ctxtual

Keep large tool results out of the context window.

When a tool returns 10,000 rows or a 500 KB page, clients truncate it and hope the answer wasn't in the part they cut. ctxtual stores the result instead and gives the model a short notice — a preview, the fields, the size — plus tools to search, filter, aggregate and page through the stored data. Small results pass through untouched.

It works two ways:

• With any MCP server, no code: ctxtual install <client> wraps the servers your AI client already uses.
• As a Python library for your own agent tools (zero dependencies).

Use it with your MCP servers

Requires uv. Wrap every server a client already has (a backup is made; uninstall puts the original entries back):

uvx --from 'ctxtual[mcp]' ctxtual install claude-desktop   # or: cursor, vscode, windsurf,
                                                           #     codex, claude-code, gemini

Restart the client. That's it. ctxtual status shows what is wrapped, --dry-run shows the changes first, and you can pass server names to wrap only some of them. (To type plain ctxtual … instead of the uvx prefix, install it once with uv tool install 'ctxtual[mcp]'.)

Adding a new server, already wrapped:

uvx --from 'ctxtual[mcp]' ctxtual add cursor github https://api.githubcopilot.com/mcp/ \
    -H 'Authorization: Bearer ${GITHUB_TOKEN}'
uvx --from 'ctxtual[mcp]' ctxtual add claude-desktop fs -- npx -y @modelcontextprotocol/server-filesystem ~/docs

add --print … prints the config snippet instead of writing it, plus a one-click install link for Cursor and VS Code (ctxtual's own options go before --; everything after it is the server's command). Or write the entry yourself — it is a normal stdio server that runs ctxtual wrap:

{
  "mcpServers": {
    "deepwiki": {
      "command": "uvx",
      "args": ["--from", "ctxtual[mcp]", "ctxtual", "wrap", "https://mcp.deepwiki.com/mcp"]
    }
  }
}

What the model sees. DeepWiki's page for the MCP Python SDK is 519,810 characters (roughly 130k tokens). Through ctxtual the model gets a 1,455-character notice:

{
  "stored_result": "read_wiki_contents_83afe435931d",
  "tool": "read_wiki_contents",
  "chars": 519810,
  "preview": "# Page: Overview …",
  "how_to_explore": "Too large to show in full (519,810 chars); stored as workspace 'read_wiki_contents_83afe435931d'. Explore it instead of calling read_wiki_contents again: result_search_in_text(workspace_id, query) finds passages; result_read_page(workspace_id, page) reads it page by page. Kept for 1 hour."
}

and then calls result_search_in_text, which returns only the matching passages. Lists get fields, a 3-item preview, and result_search / result_pipe (filter, sort, group and count in one call).

Where it helps most. Most MCP clients put a tool result into the model's context as it is, or cut it off at a limit (Codex, for example, has a per-tool token budget). That is where ctxtual makes the difference, and in agent loops you build yourself. Claude Code already saves results over 50,000 characters to files; ctxtual adds a lower threshold and query tools there. Web-only clients (claude.ai in the browser, ChatGPT) can't run local commands, so they need a hosted wrapper: see the docs.

Details — every option, OAuth, remote/hosted use, security: docs/mcp.md.

On your own FastMCP server

from fastmcp import FastMCP
from ctxtual.mcp import Offload

mcp = FastMCP("my-server")
mcp.add_middleware(Offload())          # results > 8,000 chars are stored

The Python library

For tools you call from your own agent loop, without MCP:

pip install ctxtual

Zero dependencies (standard library only), Python 3.11+.

Contents

• How it works
• Quick start
• Producers
• Built-in tools
• Pipelines
• Consumers: derived workspaces
• Dispatching tool calls
• Sessions: isolating conversations
• Keeping the tool list short
• Custom tools
• Storage backends
• Framework integrations
• Limitations
• API reference
• Development

How it works

@ctx.producer(workspace_type="papers", toolsets=[paginator(ctx), text_search(ctx)])
def search_papers(query: str) -> list[dict]:
    return database.search(query)          # 10,000 rows → stored, not returned

Calling the producer (directly or as a tool call from the model) returns:

{
  "status": "workspace_ready",
  "workspace_id": "papers_3f8a0c12b9d4",
  "item_count": 10000,
  "data_shape": "list",
  "item_schema": {"type": "object", "properties": {"title": {"type": "string"},
                  "year": {"type": "integer"}, ...}},
  "available_tools": ["papers_paginate(workspace_id='papers_3f8a0c12b9d4')", ...],
  "next_steps": ["papers_paginate: Return a page of items from the workspace...", ...]
}

The model then calls papers_search, papers_paginate, … with that workspace_id. Every exploration tool caps its output and says so when it does (has_more, truncated, total), and dispatch_tool_call rejects any result larger than Ctx(max_result_chars=60_000) — custom tools included — so no single call can flood the context.

ctxtual is a library, not a framework: it doesn't own your agent loop or your LLM calls. You export tool schemas with ctx.get_tools() and route the model's tool calls through ctx.dispatch_tool_call().

Quick start

from ctxtual import Ctx, MemoryStore
from ctxtual.utils import filter_set, paginator, pipeline, text_search

ctx = Ctx(store=MemoryStore())

@ctx.producer(workspace_type="papers", toolsets=[
    paginator(ctx),
    text_search(ctx, fields=["title", "abstract"]),
    filter_set(ctx),
    pipeline(ctx),
])
def search_papers(query: str, limit: int = 10_000) -> list[dict]:
    """Search the paper database."""
    return database.search(query, limit)

ref = search_papers("machine learning")
ws = ref["workspace_id"]

ctx.dispatch_tool_call("papers_paginate",  {"workspace_id": ws, "page": 0, "size": 10})
ctx.dispatch_tool_call("papers_search",    {"workspace_id": ws, "query": "transformer"})
ctx.dispatch_tool_call("papers_filter_by", {"workspace_id": ws, "field": "year",
                                            "value": 2024, "operator": "gte"})

# Or several operations in one call:
ctx.dispatch_tool_call("papers_pipe", {
    "workspace_id": ws,
    "steps": [
        {"search": {"query": "transformer"}},
        {"filter": {"year": {"$gte": 2024}}},
        {"sort": {"field": "citations", "order": "desc"}},
        {"limit": 5},
    ],
})

In an agent loop you pass ctx.get_tools() to the model and send each tool call to ctx.dispatch_tool_call(name, arguments); see Framework integrations. ctx.system_prompt(preamble=...) generates a short system-prompt section describing the registered tools.

Producers

@ctx.producer(
    workspace_type="papers",            # logical data type
    toolsets=[pager, search],           # tools offered for the stored data
    key="papers_{query}",               # deterministic id (optional)
    transform=lambda rows: rows[:5000], # pre-store transform (optional)
    meta={"source": "arxiv"},           # metadata shown to the model
    ttl=1800,                           # seconds until expiry (optional)
    notify=True,                        # True: dict, False: WorkspaceRef
)
def search_papers(query: str) -> list[dict]: ...

• Workspace ids. By default each call gets a random id. With key (a template over the function's parameters, or a callable receiving them), the same arguments map to the same workspace, which is replaced on each call. A key never overwrites a workspace created by a different producer or of a different type — that returns a conflict error instead. Template fields are checked against the function signature when the decorator runs.
• TTL is an absolute lifetime measured from when the data was stored; re-running the producer resets it. Call ctx.sweep_expired() periodically to delete expired workspaces (expired ones are also dropped when accessed).
• Shapes. The payload's shape (list, dict or scalar) is recorded, and the returned description only lists tools that accept that shape — a producer that returns a string gets text tools, not pagination tools.
• item_schema is inferred from up to 100 items, so the model knows field names and types before it reads any data.
• Size limit. Ctx(max_items=...) rejects larger payloads.
• Producers can be async def.

Built-in tools

Factories in ctxtual.utils. Each takes the Ctx and an optional name; when the name is omitted it is taken from the producer's workspace_type, so paginator(ctx) on a "papers" producer creates papers_paginate, etc.

All tools take workspace_id as their first argument.

Output caps (factory keyword arguments, with defaults):

Search. Terms are matched individually (any term matches; items matching more terms rank higher). On SQLiteStore, plain-word queries use FTS5 with Porter stemming, prefix matching and bm25 ranking — also when fields= restricts the search. It falls back to substring scoring (what MemoryStore always uses) when FTS finds nothing, e.g. for a fragment from the middle of a word, and for terms with symbols like C++ or e-mail, which the tokenizer can't distinguish from C or e.

Text transforms. To use the list tools on text, split it first: chunk_text(text, chunk_size, overlap), split_sections(text, separator), split_markdown_sections(text) (all in ctxtual).

Pipelines

{name}_pipe runs a list of steps in one tool call; only the final result is returned.

ctx.dispatch_tool_call("papers_pipe", {
    "workspace_id": ws,
    "steps": [
        {"flatten": "tags"},
        {"group_by": {"field": "tags", "metrics": {"n": "count", "avg": "mean:citations"}}},
        {"sort": {"field": "n", "order": "desc"}},
        {"limit": 10},
    ],
})

save_as="name" stores the full result as a new workspace saved_name (of the source's type) and returns its description. Saved results live in their own saved_ namespace and can only replace earlier saved results, never a producer's workspace.

$regex is disabled by default. Patterns come from the model and run on your server with Python's backtracking regex engine, which has no timeout. pipeline(ctx, allow_regex=True) enables it with guards: patterns are checked on Python's parse tree and rejected if they nest a repeat inside a repeat ((a+)+, (.*a){12}), repeat an alternation ((a|a)+), use backreferences, or have more than 4 unbounded repeats; values are matched only up to 1,000 characters. That removes the known exponential shapes but is not a proof of safety — enable it only when you trust the prompt.

Consumers: derived workspaces

A consumer reads one workspace and can store a new one. Consumers are registered with the Ctx: they are exported by get_tools() and callable through dispatch_tool_call().

from ctxtual import ConsumerContext

@ctx.consumer(
    workspace_type="raw",            # input must be a "raw" workspace
    produces="clean",                # output workspace type
    produces_toolsets=[paginator(ctx)],
    meta={"stage": "cleaning"},
)
def clean(workspace_id: str, min_quality: float = 0.5, *, cc: ConsumerContext):
    """Drop low-quality rows and normalize the rest."""
    rows = cc.get_items()
    return cc.emit([normalize(r) for r in rows if r["quality"] >= min_quality])

The ConsumerContext parameter (named ctx or annotated ConsumerContext) is injected and hidden from the model's schema. Calls with a missing, expired, or wrong-type workspace return an error before your function runs. Pass expose=False for consumers that only your own code should call.

Dispatching tool calls

result = ctx.dispatch_tool_call("papers_search", {"workspace_id": ws, "query": "attention"})
result = await ctx.adispatch_tool_call("papers_search", {...})   # async tools

arguments may be a dict or a JSON string. Mistakes the model can fix come back as a dict with error and suggested_action instead of an exception:

Set Ctx(safe=False) (producers/consumers) or ToolSet(safe=False) to let exceptions propagate instead — useful in tests.

The sync dispatch_tool_call runs async tools with asyncio.run when no event loop is running, and raises RuntimeError inside a running loop — use adispatch_tool_call there.

Sessions: isolating conversations

Workspace ids are the only handle the model has on stored data. If one Ctx and store serve several users or conversations, scope each to a session:

ctx.dispatch_tool_call(name, args, session_id=conversation_id)

# or for a block of code (works across async tasks, uses contextvars):
with ctx.session(conversation_id):
    search_papers("ml")

While a session is active:

• new workspaces belong to it;
• workspaces of other sessions behave exactly like missing ones — they can't be read, listed in error messages, or overwritten;
• deterministic ids (key=, save_as, emit(workspace_id=...)) get a session-derived suffix (notes_ml@3fa2c1d0e9b7), so two sessions using the same key never collide;
• workspaces created outside any session are shared and readable by all.

Without a session everything is visible, which is what single-user scripts want. Sessions are isolation, not authentication: deriving session_id from an authenticated user or conversation is up to your application.

Keeping the tool list short

Each factory adds 2–4 tools per workspace type, and models choose tools less reliably as the list grows. If you have several data sources, share one generic set of tools across all of them:

data_tools = [
    paginator(ctx, "data", enforce_type=False),
    pipeline(ctx, "data", enforce_type=False),
]

@ctx.producer(workspace_type="papers", toolsets=data_tools)
def search_papers(query: str) -> list[dict]: ...

@ctx.producer(workspace_type="people", toolsets=data_tools)
def search_people(name: str) -> list[dict]: ...

That is 6 exploration tools in total (data_paginate, data_pipe, …) no matter how many producers you add. pipeline alone covers search, filter, sort and aggregation.

Custom tools

analytics = ctx.toolset("transactions")
analytics.data_shape = "list"

@analytics.tool(
    name="transactions_anomalies",
    output_hint="Inspect one with transactions_get_item(workspace_id='{workspace_id}', index=I).",
)
def detect_anomalies(workspace_id: str, std_threshold: float = 2.0) -> dict:
    """Flag transactions whose amount deviates from the mean.

    Args:
        workspace_id: The transactions workspace to analyze.
        std_threshold: Standard deviations from the mean to flag.
    """
    items = analytics.store.get_items(workspace_id)
    ...

• The wrapper validates the workspace (existence, session, expiry, type, shape), records the access, and converts exceptions to error dicts.
• Schemas are generated from type hints and Google- or Sphinx-style docstrings; schema_extra={"param": {...}} merges extra JSON Schema.
• output_hint wraps results as {"result": ..., "_hint": "..."}.
• data_shape= on @tool overrides the toolset's shape per tool.
• async def tools are supported.
• toolset.bind(workspace_id) returns callables with the id pre-filled.

Storage backends

MemoryStore(max_workspaces=None) — in-process, thread-safe. With max_workspaces, the least recently used workspace is evicted.

SQLiteStore(path) — persistent, standard-library sqlite3, WAL mode, one connection per thread (connections of exited threads are closed). Every write runs in a transaction that is rolled back on error. List items are stored one row each, so paging, filter_by and sort_by run as SQL. Search uses an FTS5 index. Databases created by older versions are migrated on open. ":memory:" gives the store a private in-memory database shared by its threads.

Custom backends subclass BaseStore and implement init_workspace, drop_workspace, get_meta, list_workspaces, set, get and delete_key. Everything else (paging, search, filter, sort, mutations, touch, sweep_expired) has a default implementation you can override for speed. init_workspace must fully replace existing metadata.

Mutations — store.append_items, update_item, patch_item, delete_items — keep counts and the search index consistent. store.transaction() groups writes atomically on SQLite.

Framework integrations

The adapters don't import any SDK; they work with SDK objects or plain dicts.

# OpenAI Chat Completions
from ctxtual.integrations.openai import handle_tool_calls, has_tool_calls, to_openai_tools

response = client.chat.completions.create(model=MODEL, messages=messages,
                                          tools=to_openai_tools(ctx))
if has_tool_calls(response):
    messages.append(response.choices[0].message)
    messages.extend(handle_tool_calls(ctx, response, session_id=conversation_id))

# Anthropic Messages
from ctxtual.integrations.anthropic import handle_tool_use, has_tool_use, to_anthropic_tools

response = client.messages.create(model="claude-sonnet-5-5", max_tokens=4096,
                                  tools=to_anthropic_tools(ctx), messages=messages)
if has_tool_use(response):
    messages.append({"role": "assistant", "content": response.content})
    messages.append({"role": "user",
                     "content": handle_tool_use(ctx, response, session_id=conversation_id)})

# LangChain (requires langchain-core)
from ctxtual.integrations.langchain import to_langchain_tools
tools = to_langchain_tools(ctx, session_id=conversation_id)   # supports invoke and ainvoke

Async variants: ahandle_tool_calls, ahandle_tool_use.

Limitations

• Pipelines and aggregation run in Python over the whole workspace, on both backends. Fine for thousands to low hundreds of thousands of rows; for more, push work into your producer or a custom tool.
• MemoryStore.get_items returns stored objects, not copies; don't mutate them in custom tools.
• SQLiteStore serialises access with one lock, so it suits a single process. Several processes can share a database file, but there is no cross-process locking beyond SQLite's own.
• $regex is off by default — see Pipelines.
• MemoryStore(max_workspaces=...) eviction is global: one busy session can evict another session's workspaces. Size the limit for all sessions, or use TTLs.
• Sessions are not authentication — see Sessions.

API reference

Ctx

ToolSet

WorkspaceMeta

workspace_id, workspace_type, created_at, last_accessed_at, producer_fn, producer_kwargs, item_count, ttl, data_shape, extra, session_id, is_expired.

Examples

examples/ has runnable scripts (uv run python examples/01_quickstart.py) and notebooks: quickstart, a RAG support agent, multi-step data pipelines, custom tools, pipelines, SQLite persistence and mutations, error handling, multi-agent collaboration, OpenAI and Anthropic agent loops (simulated when no API key is set), a concurrent server, and the Offload middleware on a FastMCP server (uv run --extra mcp python examples/12_mcp_offload.py).

Development

git clone https://github.com/banda-larga/ctxtual.git
cd ctxtual
uv sync --extra dev --extra mcp
uv run pytest
uv run ruff check src/ tests/

Contributions are welcome; please open an issue before large changes. See CHANGELOG.md for release notes.

Releasing

1. Bump version in pyproject.toml and __version__ in src/ctxtual/__init__.py, and add a ## X.Y.Z section to CHANGELOG.md (a test checks that the three agree).
2. Commit, then tag and push: git tag -a vX.Y.Z -m "ctxtual X.Y.Z" && git push origin vX.Y.Z.

The release workflow runs the tests, checks the tag against the version, builds, publishes to PyPI with Trusted Publishing (no token) and creates the GitHub Release with the changelog section as notes.

One-time setup: on PyPI, ctxtual → Settings → Publishing → Add a new publisher → GitHub, with owner banda-larga, repository ctxtual, workflow release.yml, environment pypi. Optionally, in GitHub Settings → Environments → pypi, add yourself as a required reviewer so every publish waits for your approval.

License

MIT