graphmind-ai (Python)

A live debugger for AI agents. Phoenix and Langfuse show you what your agent did. GraphMind attaches while it's happening.

Your instrumented app streams execution events over a local WebSocket to the GraphMind viewer, which renders the run as a live graph — and can hold execution: before an LLM step, before/after a tool call, or on error. From the viewer you then resume with continue, retry, inject (substitute a result) or abort.

Everything fails open. With no debugger attached the instrumentation is a no-op measured in microseconds; if the debugger disconnects mid-hold, every held gate auto-continues in under 100 ms.

pip install graphmind-ai

Then run the viewer (from the graphmind-ai npm CLI):

npx graphmind-ai serve

• Distribution name: graphmind-ai · import name: graphmind
• Python 3.10+, one runtime dependency (websockets), MIT licensed.
• Wire protocol v1 — byte-identical to the TypeScript client, so Python and TypeScript runs render in the same viewer.

60-second quickstart

import graphmind as gm
from openai import OpenAI

client = gm.instrument_openai(OpenAI())


@gm.tool
def search_flights(origin: str, destination: str) -> list[dict]:
    return [{"flight": "TP1234", "price": 218}]


with gm.run("book-trip"):
    response = client.chat.completions.create(
        model="gpt-5",
        messages=[{"role": "user", "content": "Cheapest VIE -> LIS next Friday?"}],
        tools=[{"type": "function", "function": {"name": "search_flights"}}],
    )
    ...

That's it. No config file, no exporter, no collector. Set a breakpoint on search_flights in the viewer, run it again, and execution stops before the function body runs — nothing is in flight, so you can sit on a breakpoint for as long as you like.

Sync and async are both first-class. Most production Python agent code is synchronous, so nothing here requires an event loop:

async def main():
    async with gm.run("book-trip"):  # same object, `async with`
        await client.chat.completions.create(...)

Under the hood the transport lives on one dedicated daemon thread with its own event loop. Your loop is never touched — no nest_asyncio, no run_until_complete, no hijacking — and sync code never needs a loop at all.

Integrations

OpenAI

import graphmind as gm
from openai import OpenAI, AsyncOpenAI

client = gm.instrument_openai(OpenAI())  # sync
aclient = gm.instrument_openai(AsyncOpenAI())  # async

Patches chat.completions.create (and .parse), and responses.create, on the instance — no library monkey-patching, no import hooks. Streaming responses are teed: your code receives exactly the provider's stream while GraphMind observes deltas. tools=[...] is pre-announced as a graph.hint so the viewer renders the tool roster before anything runs.

Anthropic

import graphmind as gm
from anthropic import Anthropic

client = gm.instrument_anthropic(Anthropic())

with client.messages.stream(model="claude-sonnet-4-5", max_tokens=1024, messages=[...]) as stream:
    for text in stream.text_stream:
        print(text, end="")

Patches messages.create (including stream=True) and messages.stream. For messages.stream the HTTP request happens in __enter__, so that is where the gate holds. The stream proxy observes both consumption styles — raw event iteration and .text_stream — and recovers final token usage from the SDK's own message snapshot either way.

LangChain / LangGraph

import graphmind as gm

handler = gm.callback_handler()  # sync chains
ahandler = gm.async_callback_handler()  # async chains / LangGraph

result = chain.invoke(payload, config={"callbacks": [handler]})
result = await graph.ainvoke(payload, config={"callbacks": [ahandler]})

Chains, LLMs, chat models, tools and retrievers become graph nodes, parented by LangChain's parent_run_id, with token streaming from on_llm_new_token.

Plain functions — where inject and retry really work

from functools import partial


@gm.tool
def search_flights(origin: str, destination: str) -> list[dict]: ...


@gm.tool  # async functions stay async
async def fetch(url: str) -> str: ...


tools = gm.wrap_tools({"search": search, "book": book})  # or a list, or one callable

# functools.partial is a normal way to bind per-run state, and it works:
# the node is named after the function underneath -> `tool:load_region`.
load_eu = gm.tool(partial(load_region, "eu"))
load_us = gm.tool(partial(load_region, "us"), name="load_us")  # ...unless you say otherwise

The node name is the callable's __name__. A functools.partial has none, so the name comes from the function it wraps; an instance of a class with __call__ is named after its class. Two partials of the same function are therefore one node — same code location — which is usually what you want; pass name= (or a key in wrap_tools({...})) when you want them apart.

Anything else: spans

with gm.span("plan", kind="chain") as span:  # `async with` too
    plan = build_plan(state)
    span.set_output(plan)

Use a span for the parts of a graph GraphMind cannot see by itself — a LangGraph node body, a hand-rolled planner loop, a retrieval step in your own framework.

Capability matrix

What each attachment point can actually do. This is measured, not aspirational: every ✅ below is covered by a test in tests/.

Why inject/retry are ❌ for callbacks. LangChain callbacks are observers: the framework ignores their return value, so nothing in a callback can substitute a chain's result. GraphMind accepts those actions, warns once, and treats them as continue. To inject or retry a result, wrap the call site — @gm.tool on the tool function, or gm.span around the code you want to replace. Same story for Anthropic's messages.stream: GraphMind cannot fabricate a provider stream object, so it holds and warns rather than lying.

Holding really holds — verified, not assumed. Against the langchain_core the suite installs (1.6 at the time of writing; the floor is 0.3): sync callbacks are invoked inline by handle_event on the executing thread, so blocking there holds the chain; async callbacks are awaited directly by ahandle_event. (A sync handler inside an async chain is dispatched to the default executor and still awaited via asyncio.gather, so it holds too — at the cost of parking a thread-pool thread per concurrent run. Prefer gm.async_callback_handler() there.) Both handlers set raise_error = True so an abort can terminate the chain; every handler body is fully guarded, so the only exception that ever escapes is GraphMind's own GraphMindAbortError.

API at a glance

Every call above also exists as a method on an explicit instance (gm.GraphMind(app=...)), which is what you want when one process debugs more than one agent.

Node identity

One node per code location; executions light it up.

Every node.finished and node.error this package emits carries its instanceId, so concurrent executions of the same logical node are never mis-attributed.

Attaching, and the kill switches

The transport is lazy: it connects on first use with a 300 ms budget, then retries in the background every 10 s. An agent that starts instantly can therefore run past its first gate before the handshake lands. When you want pause guarantees from the very first event:

gm.ready(timeout=2.0)  # blocks; True once breakpoints are armed
await gm.ready_async(timeout=2.0)  # async twin

ready() never raises. False means "carry on detached" — it is not an error.

gm.configure(
    app="support-agent",  # name shown in the viewer
    url="ws://127.0.0.1:4747/ingest",
    meta={"git_sha": SHA},
)

"Production-looking" is a deliberately boring, documented rule: the first variable that is set out of GRAPHMIND_ENV, ENVIRONMENT, APP_ENV, PYTHON_ENV, ENV, DJANGO_ENV, FLASK_ENV, NODE_ENV decides, and it counts as production when its value is production or prod (case-insensitive). No hostname sniffing, no cloud metadata probes — a debugger that turns itself off for surprising reasons is worse than one you have to switch on.

A disabled session never opens a socket, never allocates a buffer, and never touches your objects: instrument_openai returns the client untouched.

Overhead

Measured by tests/test_overhead.py — median of seven runs on an Apple-silicon laptop, CPython 3.13.15, 2 000 iterations per wrapped call and 20 000 for the bare gate check. CPython 3.12.14 lands within run-to-run noise of these numbers; anything below 3.10 is unsupported and untested.

Reproduce them yourself: make install && .venv/bin/python -m pytest tests/test_overhead.py -s prints exactly the lines above. (If your default python3 predates 3.10, point the venv at a supported interpreter — make install PY=python3.13.)

The suite asserts budgets of 20 µs / 1 ms / 20 µs respectively — deliberately loose, because CI runners are noisy — so a regression that puts real work on the hot path fails CI without the budgets flapping on a slow machine.

Fail-open guarantees

• Never raises into your app. Internal failures degrade to a rate-limited warning on stderr and uninstrumented behaviour. Your own exceptions propagate untouched.
• Disconnect auto-continues. Killing the viewer mid-hold releases every held gate with continue in well under 100 ms (asserted in tests/test_failopen.py). Blocked threads also poll every 250 ms as a belt-and-braces backstop, so a held gate can never outlive the debugger.
• Interpreter exit auto-continues. An atexit hook releases held gates, and the transport thread is a daemon, so GraphMind can never keep a process alive.
• Bounded memory. Events emitted while detached go into a ring buffer (default 2000) and are replayed, oldest first with their original seq, when a viewer attaches — the viewer deduplicates.
• Bounded payloads. Prompts, tool arguments and results are depth-, width- and length-capped before serialization, so a vision agent's base64 images cannot melt the socket. Anything unserializable degrades to a bounded repr.
• fork()-safe. The loop thread is re-created in the child, so pre-forking servers (gunicorn, uvicorn workers, Celery) keep working.
• Ctrl-C works while a gate is held.

Limitations

• inject / retry are unavailable at observer-only attachment points — LangChain callbacks and Anthropic's messages.stream. See the capability matrix. Wrap the call site to get them.
• No mid-stream gates. A streamed response is observed, not pausable, once it has started. The gate is at the start of the call (matching the TypeScript adapter's documented behaviour for streaming tools).
• LangChain child-config propagation is LangChain's. A manual .ainvoke() made inside an async lambda body inherits no run config, so that child produces no callbacks — for any handler, not just this one. Compose with | or pass config= explicitly. Pinned by a test so this note stays honest.
• Thread hand-offs lose the run context. Run context lives in a contextvars.ContextVar, which propagates to asyncio tasks but not across a bare ThreadPoolExecutor.submit. Use contextvars.copy_context().run(...), or open a gm.run(...) inside the worker.
• instrument_* patches an instance. Clients created after the call are not instrumented; call it on each client you build. It is idempotent, so calling it twice is safe.
• Streaming usage needs the provider to send it. For OpenAI chat streams, pass stream_options={"include_usage": True} or the node shows no token counts.
• No provider-side timeout neutralization yet. The TypeScript adapter neutralizes SDK timeout configs while attached; the Python SDKs' timeouts are still live, so a long hold can trip a client-side timeout after the gate releases. Remove aggressive timeout= settings while debugging.
• CrewAI / LlamaIndex are not yet instrumented directly. Both run on top of provider clients, so instrument_openai / instrument_anthropic plus gm.span already give you a usable graph today.

Not ours: the GeneratorExit traceback at loop shutdown

If you stream from AsyncOpenAI, you may see this printed after your program has finished, on the way out of asyncio.run(...):

an error occurred during closing of asynchronous generator
<async_generator object PoolByteStream.__aiter__ at 0x…>
  File ".../httpcore2/_async/http11.py", line 313, in __aiter__
    yield chunk
GeneratorExit
...
RuntimeError: generator didn't stop after athrow()

It looks alarming and it names none of your code, so it is easy to blame the debugger. It is not GraphMind. openai>=3 ships httpcore2, whose connection-pool async generator is still open when asyncio.run calls loop.shutdown_asyncgens(); the generator re-raises while being thrown into, and contextlib reports that. Verified on CPython 3.12.14 and 3.13.15 with openai 3.5.0 / httpcore2 2.12.0: the same traceback appears, unchanged, with GraphMind attached, with GRAPHMIND_DISABLED=1, and with GraphMind not installed at all. The exit code is 0 and your response arrived in full.

The workaround is to give the pool one real tick to finalize itself before the loop closes:

async def main() -> None:
    client = gm.instrument_openai(AsyncOpenAI())
    try:
        ...
    finally:
        await client.close()
        await asyncio.sleep(0.05)   # let httpcore2 finalize its own generator

asyncio.run(main())

await asyncio.sleep(0) is not enough — measured; a zero-length sleep still leaves the traceback. Any small non-zero sleep clears it (1 ms was enough here); the python-analyst sample uses 0.25 s for margin.

GraphMind adds nothing to that shutdown path of its own: the stream tee is a plain class-based proxy, never an async def … yield generator, so shutdown_asyncgens() has nothing of ours to finalize (pinned by tests/test_openai.py::test_the_async_tee_adds_no_async_generator_to_your_loop). GraphMind's own transport lives on a separate daemon thread with its own loop and is never touched by your loop's shutdown.

Development

make install     # venv + dev dependencies (editable install)
make test        # pytest
make lint        # ruff check + ruff format --check
make typecheck   # mypy
make check       # all of the above
make build       # wheel + sdist into dist/
make clean

The test suite needs no API keys and no network: provider calls run through the real SDKs against an httpx.MockTransport, and the only socket is a loopback WebSocket to a fake viewer that speaks the real protocol. Emitted frames are validated against packages/schema/schema.json — the same artifact the CLI, the viewer and the TypeScript client are built from.

License

MIT.