wenchang

wenchang/integrate-without-mcp.md View as Markdown (for AI agents; also llms.txt)

Integrate without MCP#

The memory tools are methods on a Python object. If your agent runs on another tool framework, or on your own loop over a model API, call them directly. Nothing is registered anywhere until you register it.

Install#

The base package is enough. It needs no mcp:

pip install "wenchang @ git+https://github.com/msiebert/wenchang@v0.1.1"

With uv: uv add "wenchang @ git+https://github.com/msiebert/wenchang@v0.1.1".

The tools are methods#

wenchang.tools.bind_tools(client, resolver, credentials, policy, *, source, product=None, on_call=None) resolves the caller's identity and returns a MemoryTools for that identity. On it:

The methods take the same arguments as the tools in the reference and return result objects. Nothing else is exposed: no prompt, no registration, no transport.

If you have already resolved the identity yourself, MemoryTools(client, identity, policy, *, source, product=None, on_call=None) builds the same object from an Identity.

Bind per request#

bind_tools calls your resolver once and fixes that identity for the object's lifetime. A multi-tenant service binds per request, or per session if a session has one caller, never once per process. Binding runs the resolver and builds a small object.

The store, the client, the resolver, and the policy are built once per process and shared; Threads lists what is safe to share.

If the resolver returns a ResolutionFailure or raises, bind_tools raises ResolverFailureError (permanent). Render it like any tool error. The model then has no memory for that request. If resolve is async, bind_tools raises TypeError before it runs the resolver.

The multi-tenant resolver on Identity works here unchanged. Build its Credentials from your request and pass them to bind_tools; The same resolver without MCP shows the call.

Tool schemas#

wenchang.schemas.tool_schemas(product=None) returns a read-only mapping from tool name to ToolSchema, one per tool, in TOOL_NAMES order. Give these to your framework instead of writing argument models by hand. error_schema() returns the schema shared by every render_error payload.

Field Type Contents
name string The bare tool name, such as read_file.
description string The tool's description, the same as wenchang.tools.tool_descriptions(product)[name].
input dict JSON Schema of the tool's arguments, with defaults for optional ones.
output dict JSON Schema of the render_result payload on success.
error dict JSON Schema of the render_error payload on failure; the same as error_schema().

Every schema is JSON Schema draft 2020-12 and closed (additionalProperties: false). Each call builds fresh dicts, so you can change them without affecting later calls.

Validate the model's arguments against input before calling the method, as most frameworks do from the parameter schema. The method is called with **arguments, so a missing or unknown argument raises a TypeError that render_error reports as an opaque internal error. Here ToolDef stands in for your framework's tool definition, for one session's tools:

from collections.abc import Callable
from dataclasses import dataclass

from wenchang.core import MemoryStore
from wenchang.identity import Identity, SandboxResolver, ScopeGrant
from wenchang.schemas import tool_schemas
from wenchang.scope import ScopePolicy
from wenchang.storage.memory import InMemoryStorage
from wenchang.tools import MemoryTools, bind_tools, render_error, render_result
from wenchang.transport import InProcessClient


@dataclass
class ToolDef:  # stand-in for your framework's tool definition
    name: str
    description: str
    parameters: dict[str, object]
    handler: Callable[[dict[str, object]], dict[str, object]]


def handler(tools: MemoryTools, name: str) -> Callable[[dict[str, object]], dict[str, object]]:
    method = tools.tools()[name]

    def call(arguments: dict[str, object]) -> dict[str, object]:
        # The framework has already validated `arguments` against `s.input`.
        try:
            return render_result(method(**arguments))  # matches the schema's output
        except Exception as exc:
            return render_error(exc)  # matches the schema's error

    return call


client = InProcessClient(MemoryStore(InMemoryStorage()))
resolver = SandboxResolver(Identity(grants={"user": ScopeGrant("u-tianlong", "owner")}))
tools = bind_tools(client, resolver, None, ScopePolicy({}), source="inkwell-cli", product="Inkwell")

defs: dict[str, ToolDef] = {}
for s in tool_schemas(product="Inkwell").values():
    call = handler(tools, s.name)
    defs[s.name] = ToolDef(name=s.name, description=s.description, parameters=s.input, handler=call)

# A NotFoundError payload, valid against tool_schemas()["read_file"].error.
print(defs["read_file"].handler({"scope": "user", "area": "notes", "name": "tea"}))

The schemas assume on-contract output and standard tool-use modes:

Results and errors#

Tool methods return result objects and raise exceptions. wenchang.tools.render_result(value) turns a return value into the JSON-safe dict to show the model, and render_error(exc) turns any exception into the error payload, with the category and repair details described under errors. These are the same payloads the MCP adapter sends. An unexpected exception renders as an internal error with a generic message, so internals never reach the model.

A complete example#

This binds per request with an in-memory store, dispatches each call by name with arguments checked against the schema, and renders a result and an error. The asserts show what comes back.

from wenchang.core import MemoryStore
from wenchang.errors import InvalidArgumentError
from wenchang.identity import Identity, ResolutionFailure, ScopeGrant
from wenchang.observer import ToolCallRecord
from wenchang.schemas import tool_schemas
from wenchang.scope import ScopePolicy
from wenchang.storage.memory import InMemoryStorage
from wenchang.tools import bind_tools, render_error, render_result
from wenchang.transport import InProcessClient

USERS = {"tok-tianlong": "u-tianlong", "tok-diya": "u-diya"}  # stand-in for your IdP


class TokenResolver:
    """Maps a bearer token to the caller's grants. Must not raise."""

    def resolve(self, credentials: str | None) -> Identity | ResolutionFailure:
        user_id = USERS.get(credentials or "")
        if user_id is None:
            return ResolutionFailure("Invalid or expired token.")
        return Identity(grants={"user": ScopeGrant(user_id, "owner")})


# Built once per process and shared by every request. InMemoryStorage is safe to share
# across threads, as is a GcsStorage; see the Threads section.
client = InProcessClient(MemoryStore(InMemoryStorage()))
resolver = TokenResolver()
policy = ScopePolicy({})
schemas = tool_schemas(product="Inkwell")
records: list[ToolCallRecord] = []

# Register these with your framework once: name, description, and parameter schema.
definitions = [
    {"name": s.name, "description": s.description, "input_schema": s.input}
    for s in schemas.values()
]


def check_arguments(name: str, arguments: dict[str, object]) -> None:
    # Your framework normally validates against the input schema; this checks names only.
    schema: dict = schemas[name].input
    for key in arguments:
        if key not in schema["properties"]:
            raise InvalidArgumentError(key, "not a parameter of this tool")
    for key in schema.get("required", []):
        if key not in arguments:
            raise InvalidArgumentError(key, "required")


def run_tool_calls(
    token: str | None, calls: list[tuple[str, dict[str, object]]]
) -> list[dict[str, object]]:
    """Run one request's tool calls as the caller the token names."""
    try:
        # Resolve once per request, never once per process.
        tools = bind_tools(
            client,
            resolver,
            token,
            policy,
            source="inkwell-api",
            product="Inkwell",
            on_call=records.append,
        )
    except Exception as exc:  # ResolverFailureError: no memory for this request
        return [render_error(exc) for _ in calls]
    methods = tools.tools()
    payloads: list[dict[str, object]] = []
    for name, arguments in calls:
        try:
            check_arguments(name, arguments)
            payloads.append(render_result(methods[name](**arguments)))
        except Exception as exc:
            payloads.append(render_error(exc))
    return payloads


created, missing = run_tool_calls(
    "tok-tianlong",
    [
        (
            "write_file",
            {
                "scope": "user",
                "area": "preferences",
                "name": "prose-style",
                "content": "- [stated] Writes in close third person, past tense.\n",
                "description": "Prose style preferences",
                "aliases": ["narration", "voice"],
                "expected_version": None,
            },
        ),
        ("read_file", {"scope": "user", "area": "preferences", "name": "tea"}),
    ],
)
assert created["name"] == "prose-style" and created["sources"] == ["inkwell-api"]
assert missing["error"] == "NotFoundError" and missing["category"] == "recoverable"

[denied] = run_tool_calls("tok-expired", [("get_memory_index", {})])
assert denied["error"] == "ResolverFailureError" and denied["category"] == "permanent"

# One record per tool call; request_id is None for direct calls.
assert [(r.tool, r.is_error, r.request_id) for r in records] == [
    ("write_file", False, None),
    ("read_file", True, None),
]
print(created["version"], missing["message"])

Threads and async hosts#

Calls are synchronous. Under GcsStorage each one blocks on one or more GCS round trips; get_memory_index lists every object in the caller's entities unless an index cache serves it. This section applies to the MCP paths too: the MCP adapter runs tool calls concurrently on worker threads and calls one client from all of them.

What can be shared#

Object Across threads
MemoryStore Safe. It sets its fields once in the constructor and never changes them. All its state lives in its storage and its index cache.
InProcessClient Safe. It holds only its store.
MemoryTools Safe; it never changes after construction. It holds one caller's identity, so share it only within that caller's request.
InMemoryStorage, InMemoryIndexCache Safe. Each guards its state with a lock.
FilesystemStorage Safe. A per-instance lock and a file lock on root/.lock serialize operations, so instances and processes on one root are safe too.
GcsStorage Safe. It adds no state of its own: it holds the bucket and builds a new Blob for every call. Its constructor builds the client's HTTP session on the constructing thread.
google.cloud.storage.Client and its Bucket Safe, with the caveats below. One client per process works; heavy concurrency can outgrow its connection pool.
Your resolver, IndexCache functions, on_call, and clock Yours. If tool calls run on several threads, as they always do over MCP, each must be safe to call from several threads.

The index cache's generation tokens already handle readers and writers racing on one entity, across threads and processes; your get and set need only be safe to call concurrently and linearizable per key (Guarantees and limits).

One shared storage.Client is safe. Build one per process and share it, its Bucket, and one GcsStorage across threads; it needs no extra clients and no locking. The basis is a reading of google-cloud-storage 3.14.1, google-auth, and requests, plus wenchang's integration tests, which stress one shared client against an emulator and re-verify this on upgrades. It is verified against google-cloud-storage 3.14.1, the version wenchang's lockfile resolves; older versions in the declared range (>=2.18) are not verified. In the source, the batch stack is thread-local, the bucket metadata cache is lock-guarded, and each request keeps its state in its own arguments and copied headers. The library's own transfer_manager shares one client across threads, which is an internal precedent rather than an API contract. GcsStorage(bucket) builds the client's HTTP session when it is constructed, so no request runs before the session is fully set up. The caveats:

Async hosts#

An async host must not call wenchang on its event loop. Run each tool call in a worker thread, for example await asyncio.to_thread(method, **arguments). Call bind_tools in a worker thread too, tools = await asyncio.to_thread(bind_tools, client, resolver, credentials, policy, source="inkwell-api"), because it runs your resolver, which blocks on whatever verification it does. The resolver itself must be a plain synchronous method; bind_tools raises TypeError for an async def resolve (Identity).

Observing calls#

bind_tools(..., on_call=...) delivers one ToolCallRecord per tool call. On this path:

The observer runs synchronously on the calling thread, before the call returns or raises, so correlate records with a contextvar or thread-local. bind_tools checks on_call before it runs the resolver. A call Python rejects for its arguments never produces a record. An exception the observer raises is logged at ERROR and swallowed. Records contain memory content, so treat them as sensitive. The reference lists every field.

This observer logs and counts calls without recording arguments or results:

import logging
import threading
from collections import Counter, defaultdict
from contextlib import suppress

from wenchang.core import MemoryStore
from wenchang.errors import NotFoundError
from wenchang.identity import Identity, SandboxResolver, ScopeGrant
from wenchang.observer import ToolCallRecord
from wenchang.scope import ScopePolicy
from wenchang.storage.memory import InMemoryStorage
from wenchang.tools import bind_tools
from wenchang.transport import InProcessClient

log = logging.getLogger("inkwell.memory")

# Stand-ins for your metrics client: a call counter and per-tool latencies.
lock = threading.Lock()
calls: Counter[tuple[str, str]] = Counter()
latencies: defaultdict[str, list[float]] = defaultdict(list)


def on_call(record: ToolCallRecord) -> None:
    # arguments and result carry memory content; log only the call's shape.
    category = record.result.get("category") if record.is_error else None
    log.info(
        "memory tool=%s is_error=%s category=%s duration_s=%.3f",
        record.tool,
        record.is_error,
        category,
        record.duration_s,
    )
    with lock:  # under MCP, observers run on worker threads
        calls[(record.tool, str(category or "ok"))] += 1
        latencies[record.tool].append(record.duration_s)


client = InProcessClient(MemoryStore(InMemoryStorage()))
resolver = SandboxResolver(Identity(grants={"user": ScopeGrant("u-tianlong", "owner")}))
tools = bind_tools(client, resolver, None, ScopePolicy({}), source="inkwell-cli", on_call=on_call)

tools.get_memory_index()  # on_call has run by the time this returns
with suppress(NotFoundError):  # on_call sees it first: is_error=True, category=recoverable
    tools.read_file("user", "preferences", "csv")
print(calls)

What you provide#