Conversation

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

def conversation(self, toolkit: Toolkit, cancel: asyncio.Event | None = None) -> Conversation

class Conversation:
    messages: list[dict[str, Any]]
    async def send(self, prompt: str) -> RunResult: ...
    def reset(self) -> None: ...

An in-process, in-memory handle onto one multi-turn conversation. Each send() calls Client.run with history=self.messages and stores the updated transcript back onto self.messages, so the next send() continues where the last one left off.

When to use it

• Interactive, single-process use — a REPL, a CLI session, a WebSocket handler holding one live Conversation per connection — where the process itself is the natural lifetime of the memory.
• You want the object to carry the transcript (convo.messages), rather than threading history= through run() calls yourself, or looking a conversation up by id from a store on every turn.

Why this and not the alternative

Reach for a ConversationStore when memory must outlive the process

Conversation holds messages in a plain Python list — gone when the process exits. Client.ask(prompt, toolkit, id=...) is the store-backed equivalent: the same append-and-continue behavior, but persisted through a ConversationStore so a conversation survives a restart or is shared across processes. Use Conversation for same-process memory; ask(..., id=...) for durable memory.

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 — two turns, growing transcript

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": 1, "completion_tokens": 1, "total_tokens": 2},
    }


async def main():
    tk = await create_toolkit()
    try:
        with StubServer(reply) as srv:
            client = create_client(base_url=srv.base_url, style="openai", model="test-model", api_key="test-key")
            convo = client.conversation(tk)

            await convo.send("first")
            n1 = len(convo.messages)
            await convo.send("second")
            n2 = len(convo.messages)

        assert n2 > n1, "the transcript must grow with each send()"
        users = [m["content"] for m in convo.messages if m.get("role") == "user"]
        assert users == ["first", "second"]

        print("ok:", n1, "->", n2, "messages")
    finally:
        await tk.close()


asyncio.run(main())

2. The realistic case — the model recalls a fact from turn 1

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()


async def main():
    tk = await create_toolkit()

    # A stub that "remembers" by echoing back whatever prior user turn is in the
    # transcript it's handed — this is what a real model does with real history.
    def stateful_reply(body):
        prior_users = [m["content"] for m in body["messages"] if m.get("role") == "user"]
        if len(prior_users) >= 2:
            return {"choices": [{"message": {"role": "assistant", "content": "Muthu, favorite number 7"}}], "usage": {"prompt_tokens": 2, "completion_tokens": 2, "total_tokens": 4}}
        return {"choices": [{"message": {"role": "assistant", "content": "noted"}}], "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}}

    try:
        with StubServer(stateful_reply) as srv:
            client = create_client(base_url=srv.base_url, style="openai", model="test-model", api_key="test-key")
            convo = client.conversation(tk)

            a = await convo.send("My name is Muthu and my favorite number is 7. Reply 'noted'.")
            b = await convo.send("What is my name and favorite number?")

        assert a.text == "noted"
        assert "Muthu" in b.text and "7" in b.text

        print("ok:", a.text, "->", b.text)
    finally:
        await tk.close()


asyncio.run(main())

3. The full surface — reset() clears memory, cancel token is threaded through

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

from toolnexus import RunCancelled, 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": 1, "completion_tokens": 1, "total_tokens": 2},
    }


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

            # reset(): the transcript goes back to empty, as if the conversation just started.
            convo = client.conversation(tk)
            await convo.send("first")
            assert len(convo.messages) > 0
            convo.reset()
            assert convo.messages == []
            await convo.send("after reset")
            # only "after reset" is on the transcript — "first" is gone.
            users = [m["content"] for m in convo.messages if m.get("role") == "user"]
            assert users == ["after reset"]

            # cancel: an asyncio.Event passed to conversation() is honored by every send().
            cancel = asyncio.Event()
            cancel.set()  # already fired — the very next send() aborts immediately
            convo2 = client.conversation(tk, cancel=cancel)
            try:
                await convo2.send("should not complete")
                raised = False
            except RunCancelled:
                raised = True
            assert raised

        print("ok: reset ->", users, "| cancel raised RunCancelled:", raised)
    finally:
        await tk.close()


asyncio.run(main())

Members

Member	Type	What it is
messages	list[dict]	The full running transcript — read it directly, or hand it to another run(..., history=...) call.
send(prompt)	async (str) -> RunResult	Send the next user turn; prior history is retained automatically.
reset()	() -> None	Clear the transcript — the next send() starts a fresh conversation.

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.