Observability and egress

Two questions with mechanical answers: what did a call cost, and what can the model see. Both are read from the same declarations the runtime executes, in your process, and go nowhere else.

`ToolCall`

@dataclass(frozen=True)
class ToolCall: ...

One invocation, measured. Minted on every call — successful or not, and including the ones that never reached the network.

Identity

str

The tool’s name.

Optional[str]

The provider identifier, when the tool was built by an OAuth factory.

str

HTTP method.

str

The template, never the resolved URL: the resolved path carries customer identifiers, and it would make every call its own label in a metrics backend.

Outcome

str

One of invalid_input, transform_failed, credential_unavailable, ok, envelope_error, http_error, credential_rejected, transport_error, cancelled, error — in the order a summary lists them. The first three never reached the network.

bool

Whether the tool’s own request went out. This, not the outcome name, is what counts the calls that cost nothing.

Optional[int]

The response status, when there was a response.

Optional[str]

The exception class name, when the call raised.

Three of those values exist because a status code cannot express them. invalid_input and transform_failed happened before anything was sent — the model produced a bad call and it cost no request, no quota and no money. envelope_error is the opposite surprise: HTTP 200 with the failure inside the body, which a generic HTTP metric records as a success. cancelled is separated from transport_error on purpose: a caller’s timeout is not the API failing, and filing it as one puts someone else’s latency budget in your provider’s column.

Timings

All in milliseconds, decomposed by owner.

float

Wall time for the whole invocation.

float

Deriving this tool’s view, on the one call that had to. Zero on every other.A tool’s view is built on first use rather than at import, so whichever call arrives first pays for it. On a schema whose types refer to each other that is seconds. It has its own field because it used to be reported as validate_ms, which put a one-off build in the number you watch to find out whether the model is sending malformed arguments.Tool.prepare moves it to startup. After that, a non-zero schema_ms in production means a tool nobody prepared.

float

Validating the arguments against the LLM schema. This call’s arguments only. See schema_ms for the build.

float

Building the request, including Format transforms.

float

The credential provider’s own time — reported separately so a slow vault does not read as a slow API.

float

The provider’s share: the HTTP call.

float

The response handler.

float

Derived. total_ms minus upstream_ms, floored at zero — the number that answers “is this layer in my way?”.

Sizes

In bytes, and named bytes. All four are None unless the call was observed: a sink was attached, or the charter logger was enabled for INFO. Timings are a few clock reads and always taken; sizes cost a serialisation, so they are paid for only when something will read them.

Optional[int]

The arguments as compact JSON.

Optional[int]

The request body as sent.

Optional[int]

The response as the server formatted it.

Optional[int]

The result the model receives, as compact JSON. None when the call failed.

Optional[int]

Derived. payload_bytes - context_bytes. Negative is possible and is not hidden: a handler that adds more than it removes is worth seeing.

Optional[float]

Derived. saved_bytes as a fraction of what the API sent. None when either size is missing or the payload was empty.

Bytes are not tokens. A tokens-saved figure derived from a bytes-per-token constant would be a guess with a decimal point in it, so nothing here prints one. One caveat on saved_bytes: payload_bytes is the response as the server formatted it, so against an API that pretty-prints, some of the difference is that API’s whitespace.

`ToolCall.to_dict`

def to_dict(self) -> Dict[str, Any]: ...

Flat attributes using OpenTelemetry’s names where one exists — http.request.method, http.route, http.response.status_code, peer.service, error.type — and charter.* for the rest. Keys whose value is None are dropped. Forwarding a record to an OTel span or a log pipeline needs no translation table.

`CallSink`

CallSink = Callable[[ToolCall], None]

Something to hand each finished record to. Declared as on_call on a tool or a factory. Called synchronously, once per invocation, after the call has finished and before any exception propagates. It must not be slow, and it must not raise — one that does is caught and logged at DEBUG, never allowed to fail the call it was only supposed to describe. A sink that raises does not stop the others from being called.

`collecting`

Send every call made inside a block to a sink, without attaching it to a tool.

@contextmanager
def collecting(*sinks: CallSink) -> Iterator[None]: ...

on_call is a constructor argument, and a shipped pack calls the constructor at import. charter.packs.gmail exports one set of tool objects for the whole process, so assigning on_call to one of them reaches every other caller in that process. collecting attaches a sink to a scope instead.

from charter import CallCollector, collecting, format_call_summary
from charter.auth import StaticTokenProvider
from charter.packs import gmail

gmail.configure(StaticTokenProvider("ya29..."))

calls = CallCollector()
with collecting(calls):
    await gmail.labels_list.ainvoke({"userId": "me"})

print(format_call_summary(calls))

The scope is a context, not a process. A context is copied whenever a task is created, so a sink set before gather reaches every branch, and two requests served concurrently keep their records apart without knowing about each other. It follows work off the event loop only where the copy is made for you.

Where the call happens	Sees the sink
`asyncio.create_task`, `gather`	yes
`asyncio.to_thread`	yes
`loop.run_in_executor`	no
a bare `threading.Thread`	no

In the last two, open the block inside that thread rather than around it. Nesting adds rather than replaces, and a tool’s own on_call fires too. An outer collector that stopped receiving because something inside opened its own would be a silent hole in a measurement. A sink named twice is called once.

`current_sinks`

The sinks collecting has active in this context.

def current_sinks() -> Tuple[CallSink, ...]: ...

Empty outside any collecting block. Useful for deciding whether to build an expensive record at all; the runtime uses it for exactly that, skipping payload sizing when nothing is listening.

`CallCollector`

class CallCollector:
    def __init__(self, max_records: int = 1000) -> None: ...

A CallSink that keeps what it is given, for a summary at the end. Pass it straight to a factory’s on_call.

int

default:"1000"

How many individual records to retain. Counters and totals stay exact past this; percentiles are computed over what was retained.

int

Calls seen. Also len(collector).

int

Records not retained, past max_records.

Dict[str, int]

Count per outcome.

float

Summed timings. Sums, not wall clock: concurrent calls are each counted.

int

Summed sizes over the calls that were measured.

List[ToolCall]

The retained records, oldest first. Iterating the collector yields the same.

None

Forget everything, keeping the collector usable.

Thread-safe through one lock. Not a metrics backend: it holds one process’s own run in memory and forgets it on exit.

`format_call_line`

def format_call_line(call: ToolCall) -> str: ...

One record as a single scannable line. This is the message of the INFO record the runtime emits, so enabling logging at INFO gives a call log without writing a sink:

stripe.list_charges    GET  v1/charges       200    412ms  ↑ 118 B  ↓ 38.4 KB → 412 B  (99%)
stripe.create_charge   POST v1/charges       ✗ invalid_input — not sent, 0ms

Sizes, never values. The arguments and the response body are logged only on the DEBUG path, which a host has to ask for.

`format_call_summary`

def format_call_summary(calls: Iterable[ToolCall]) -> str: ...

A run’s calls, totalled, for a human at the end of a script or a test. Accepts a CallCollector or any iterable of records; an empty one returns "charter — no calls recorded". Reports the outcome breakdown, how many calls never reached the network, context trimmed, upstream p50 and p95 (nearest-rank, over calls that reached the network), Charter’s own overhead, and a per-tool table.

from charter import CallCollector, api_key_tool_factory, format_call_summary

calls = CallCollector()
api = api_key_tool_factory(
    base_url="https://api.example.com/",
    api_key_headers={"x-api-key": "key-123"},
    on_call=calls,
)

print(format_call_summary(calls))

`egress_map`

def egress_map(tools: Iterable[Tool]) -> Dict[str, Dict[str, Any]]: ...

What each tool can and cannot expose to the model. Walks the schema every tool was declared with and splits every field, at every nesting depth, into what reaches the model and what is withheld — with the reason taken from the declaration. Returns a plain dict, keyed by tool name, so it serialises straight to JSON for an audit trail.

Dict[str, Any]

One entry per tool.

Show entry

Optional[str]

The tool’s identity, copied from the declaration.

List[str]

Dotted field paths the model can see.

List[Dict[str, str]]

One field and reason pair per hidden field. The reason is "disabled", "response_only", "mode=..." naming the modes that did not match, or "filtered".

Recursion stops at depth 12 and at a schema already seen, so a self-referential model terminates.

from charter import egress_map
from charter.packs import gmail

report = egress_map(gmail.TOOLS)
withheld = {entry["field"] for entry in report["messages_send"]["withheld"]}
assert "body.id" in withheld
assert "body.raw" in report["messages_send"]["visible"]

`conflict_map`

def conflict_map(tools: Iterable[Tool]) -> Dict[str, List[Dict[str, Any]]]: ...

The exclusion rules each tool declares, by tool name — every ConflictsWith in the pack, read back from the declarations the runtime enforces. The marker puts each rule on the field it is about, which is the right place to declare one and the wrong place to read the set of them: “what does this pack refuse together?” becomes a walk over every field rather than a glance at one list. This is that glance. Only tools that declare a conflict appear. It reads the schema the tool was declared with, so a field withheld from the model by Mode is still reported: the rule binds the request, not the prompt.

List[Dict[str, Any]]

One entry per rule.

Show entry

str

The field declaring the rule, as named in Python.

str

The same field as the API names it — a WireName when it has one, otherwise the published name.

List[Dict[str, Any]]

One field, api_name and declared entry per excluded parameter. declared is false when the rule names a field the schema does not have — such a rule never fires, and is reported rather than resolved.

str

The clause appended to the error, or "".

`format_conflicts`

def format_conflicts(tools: Iterable[Tool]) -> str: ...

conflict_map rendered for a human reading a review, grouped by what is excluded rather than by the field declaring it — because that is the question being asked.

from charter import format_conflicts
from charter.packs import gcalendar

print(format_conflicts(gcalendar.TOOLS))

events_list
  syncToken refuses 8:
    - iCalUID
    - orderBy
    - q
    ...
    why: An incremental sync continues the query the token came from, so the
         filters have to be the ones already in effect.

Returns "no exclusion rules declared" when there are none.

`format_egress_map`

def format_egress_map(tools: Iterable[Tool]) -> str: ...

egress_map rendered for a human reading a review: one block per tool, + for visible fields and - field [reason] for withheld ones.

What a call cost — the narrative version, with output worth reading
Egress control — why the map cannot drift from the runtime
Mode — the marker the map reports on

Core

Schema

Authentication

Runtime

Operating

Observability and egress

`ToolCall`

Identity

Outcome

Timings

Sizes

`ToolCall.to_dict`

`CallSink`

`collecting`

`current_sinks`

`CallCollector`

`format_call_line`

`format_call_summary`

`egress_map`

`conflict_map`

`format_conflicts`

`format_egress_map`

​ToolCall

​Identity

​Outcome

​Timings

​Sizes

​ToolCall.to_dict

​CallSink

​collecting

​current_sinks

​CallCollector

​format_call_line

​format_call_summary

​egress_map

​conflict_map

​format_conflicts

​format_egress_map

​Related

`ToolCall`

Identity

Outcome

Timings

Sizes

`ToolCall.to_dict`

`CallSink`

`collecting`

`current_sinks`

`CallCollector`

`format_call_line`

`format_call_summary`

`egress_map`

`conflict_map`

`format_conflicts`

`format_egress_map`

Related