MetricEvent

Python · package toolnexus · SPEC §8 · python/src/toolnexus/client.py

MetricEvent = dict[str, Any]
OnMetric = Callable[[MetricEvent], None]

# create_client(..., on_metric: OnMetric | None = None)
# client.metrics() -> str   # cumulative Prometheus text exposition

on_metric is a sink the client calls once per semantic event as a run happens — one "llm" event per model call, one "tool" event per tool call, one "run" event per finished run/ask/stream. The same events also feed a built-in in-memory Prometheus registry, rendered by client.metrics().

When to use it

• You want structured observability — cost/latency dashboards, per-tool error rates, alerting on retries — without instrumenting run/stream call sites yourself.
• You want a ready-made /metrics endpoint: client.metrics() renders standard Prometheus text exposition, byte-identical across all six ports, that any scraper understands.

Why this and not the alternative

on_metric is push, metrics() is pull

on_metric is called synchronously as events happen — use it to stream events to your own telemetry pipeline (structured logs, a custom sink, an in-process counter). client.metrics() is the passive alternative: nothing to wire up, just expose the string from an HTTP handler for Prometheus to scrape periodically. Use both — they’re fed by the same events, not a choice between features.

Examples

Hermetic on purpose

Every example points create_client at a local http.server stub instead of a real LLM endpoint — no network, no API key.

1. The smallest useful call — on_metric sees one llm and one run event

import asyncio
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from toolnexus import create_client, create_toolkit


class StubServer:
    def __init__(self, handler):
        outer = self

        class H(BaseHTTPRequestHandler):
            def log_message(self, *a):
                pass

            def do_POST(self):  # noqa: N802
                length = int(self.headers.get("Content-Length", 0))
                body = json.loads(self.rfile.read(length) or b"{}")
                outer._send(self, handler(body))

        self._server = ThreadingHTTPServer(("127.0.0.1", 0), H)
        self.port = self._server.server_address[1]
        self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)

    @staticmethod
    def _send(req, payload):
        body = json.dumps(payload).encode("utf-8")
        req.send_response(200)
        req.send_header("Content-Type", "application/json")
        req.send_header("Content-Length", str(len(body)))
        req.end_headers()
        req.wfile.write(body)

    @property
    def base_url(self) -> str:
        return f"http://127.0.0.1:{self.port}/v1"

    def __enter__(self):
        self._thread.start()
        return self

    def __exit__(self, *exc):
        self._server.shutdown()
        self._server.server_close()


def reply(body):
    return {
        "choices": [{"message": {"role": "assistant", "content": "ok"}}],
        "usage": {"prompt_tokens": 4, "completion_tokens": 2, "total_tokens": 6},
    }


async def main():
    tk = await create_toolkit()
    events = []
    try:
        with StubServer(reply) as srv:
            client = create_client(
                base_url=srv.base_url, style="openai", model="test-model", api_key="test-key",
                on_metric=events.append,
            )
            await client.run("hi", tk)

        kinds = [e["event"] for e in events]
        assert kinds == ["llm", "run"]

        llm_ev = events[0]
        assert llm_ev["status"] == "ok"
        assert llm_ev["prompt_tokens"] == 4 and llm_ev["completion_tokens"] == 2
        assert llm_ev["ms"] >= 0

        run_ev = events[1]
        assert run_ev["turns"] == 1 and run_ev["tool_calls"] == 0
        assert run_ev["total_tokens"] == 6

        print("ok:", kinds)
    finally:
        await tk.close()


asyncio.run(main())

2. The realistic case — a tool event alongside llm/run

import asyncio
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from toolnexus import create_client, create_toolkit, define_tool


class StubServer:
    def __init__(self, handler):
        outer = self

        class H(BaseHTTPRequestHandler):
            def log_message(self, *a):
                pass

            def do_POST(self):  # noqa: N802
                length = int(self.headers.get("Content-Length", 0))
                body = json.loads(self.rfile.read(length) or b"{}")
                outer._send(self, handler(body))

        self._server = ThreadingHTTPServer(("127.0.0.1", 0), H)
        self.port = self._server.server_address[1]
        self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)

    @staticmethod
    def _send(req, payload):
        body = json.dumps(payload).encode("utf-8")
        req.send_response(200)
        req.send_header("Content-Type", "application/json")
        req.send_header("Content-Length", str(len(body)))
        req.end_headers()
        req.wfile.write(body)

    @property
    def base_url(self) -> str:
        return f"http://127.0.0.1:{self.port}/v1"

    def __enter__(self):
        self._thread.start()
        return self

    def __exit__(self, *exc):
        self._server.shutdown()
        self._server.server_close()


def scripted(body):
    if not any(m.get("role") == "tool" for m in body["messages"]):
        return {
            "choices": [{
                "message": {
                    "role": "assistant",
                    "content": None,
                    "tool_calls": [{"id": "call_1", "type": "function", "function": {"name": "add", "arguments": '{"a": 2, "b": 3}'}}],
                },
                "finish_reason": "tool_calls",
            }],
            "usage": {"prompt_tokens": 4, "completion_tokens": 2, "total_tokens": 6},
        }
    return {"choices": [{"message": {"role": "assistant", "content": "5"}}], "usage": {"prompt_tokens": 5, "completion_tokens": 1, "total_tokens": 6}}


async def main():
    tk = await create_toolkit()
    tk.register(define_tool(lambda a, b: str(a + b), name="add", description="Add.", input_schema={
        "type": "object", "properties": {"a": {"type": "number"}, "b": {"type": "number"}}, "required": ["a", "b"],
    }))
    events = []
    try:
        with StubServer(scripted) as srv:
            client = create_client(
                base_url=srv.base_url, style="openai", model="test-model", api_key="test-key",
                on_metric=events.append,
            )
            await client.run("add 2 and 3", tk)

        kinds = [e["event"] for e in events]
        assert kinds == ["llm", "tool", "llm", "run"]

        tool_ev = next(e for e in events if e["event"] == "tool")
        assert tool_ev["tool"] == "add"
        assert tool_ev["source"] == "native"  # define_tool()'s default source
        assert tool_ev["is_error"] is False
        assert tool_ev["ms"] >= 0

        run_ev = events[-1]
        assert run_ev["tool_calls"] == 1 and run_ev["turns"] == 2

        print("ok:", kinds, "| tool source:", tool_ev["source"])
    finally:
        await tk.close()


asyncio.run(main())

3. The full surface — client.metrics() Prometheus text exposition

import asyncio
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from toolnexus import create_client, create_toolkit


class StubServer:
    def __init__(self, handler):
        outer = self

        class H(BaseHTTPRequestHandler):
            def log_message(self, *a):
                pass

            def do_POST(self):  # noqa: N802
                length = int(self.headers.get("Content-Length", 0))
                body = json.loads(self.rfile.read(length) or b"{}")
                outer._send(self, handler(body))

        self._server = ThreadingHTTPServer(("127.0.0.1", 0), H)
        self.port = self._server.server_address[1]
        self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)

    @staticmethod
    def _send(req, payload):
        body = json.dumps(payload).encode("utf-8")
        req.send_response(200)
        req.send_header("Content-Type", "application/json")
        req.send_header("Content-Length", str(len(body)))
        req.end_headers()
        req.wfile.write(body)

    @property
    def base_url(self) -> str:
        return f"http://127.0.0.1:{self.port}/v1"

    def __enter__(self):
        self._thread.start()
        return self

    def __exit__(self, *exc):
        self._server.shutdown()
        self._server.server_close()


def reply(body):
    return {
        "choices": [{"message": {"role": "assistant", "content": "ok"}}],
        "usage": {"prompt_tokens": 3, "completion_tokens": 1, "total_tokens": 4},
    }


async def main():
    tk = await create_toolkit()
    try:
        # Before any activity, metrics() is empty but valid — HELP/TYPE lines only.
        client = create_client(base_url="http://127.0.0.1:1", style="openai", model="test-model", api_key="test-key")
        empty = client.metrics()
        assert "toolnexus_llm_requests_total" in empty
        assert 'toolnexus_llm_requests_total{' not in empty  # no series recorded yet

        with StubServer(reply) as srv:
            client = create_client(base_url=srv.base_url, style="openai", model="test-model", api_key="test-key")
            await client.run("hi", tk)
            await client.run("hi again", tk)

        text = client.metrics()

        # Cumulative across BOTH run() calls: 2 successful requests for this model.
        assert 'toolnexus_llm_requests_total{model="test-model",status="ok"} 2' in text
        # Token counters and duration histograms are standard Prometheus series.
        assert "toolnexus_llm_tokens_total" in text
        assert "toolnexus_llm_request_duration_seconds_bucket" in text
        assert "# TYPE toolnexus_llm_requests_total counter" in text

        print("ok: metrics() renders", len(text.splitlines()), "lines of Prometheus text")
    finally:
        await tk.close()


asyncio.run(main())

MetricEvent shapes

event	Fields	Emitted
"llm"	model, status ("ok"|"error"), ms, prompt_tokens, completion_tokens	Once per LLM HTTP attempt that completes or fails.
"tool"	tool, source, is_error, ms, pending?	Once per tool call. pending is True iff the result is a §10 suspension — never counted as an error.
"run"	model, turns, tool_calls, total_tokens, ms, error?	Once per finished run/ask/stream. error is present only when the run raised.

See also

• create_client — The unified client: system prompt, skills injection, parallel and chained tool calls, retries, memory.
• Client.run — Send a prompt, let the loop call tools until the model stops, get a RunResult.
• Client.stream — The streaming loop: text deltas, tool-call events, and suspension events as they happen.
• Hooks — Intercept before/after model calls and tool calls: audit, redact, veto, or rewrite.