Ubiquity

A Claude-Code-style agent SDK for Python, built on pydantic-ai.

Claude Code's architecture — the agent loop, a built-in tool suite, a rule-based permission system, hooks, subagents, MCP, and session persistence — reimplemented against pydantic-ai's model layer, so the same agent runs on 604 models across 22 providers instead of one.

import asyncio
from ubiquity import query, Options

async def main():
    async for message in query(
        "what Python files are in this project?",
        Options(model="openai:gpt-5"),
    ):
        if message.type == "assistant":
            print(message.text)

asyncio.run(main())

Swap the model string and nothing else changes:

Options(model="google:gemini-3-pro")
Options(model="groq:llama-3.3-70b-versatile")
Options(model="mistral:mistral-large-latest")
Options(model="bedrock:meta.llama3-70b-instruct-v1:0")

There is no default model. Leaving model unset reads UBIQUITY_MODEL, and a run with neither configured fails with an explicit error rather than silently picking a vendor. Aliases let code name a role instead of a provider:

from ubiquity import register_alias

register_alias("fast", "groq:llama-3.3-70b-versatile")
Options(model="fast")

UBIQUITY_MODEL_ALIASES="fast=groq:llama-3.3-70b-versatile,big=openai:gpt-5" does the same from the environment.

For anything OpenAI-compatible that isn't a registered provider — Ollama, vLLM, LM Studio, OpenRouter, Together:

from ubiquity import openai_compatible

Options(
    model=openai_compatible(
        "llama3.3", 
        base_url="http://localhost:11434/v1"
    )
)

Credentials

By default a provider reads its own environment variable — GROQ_API_KEY, ANTHROPIC_API_KEY, CO_API_KEY, and so on. To pass a key per run instead:

Options(model="groq:openai/gpt-oss-120b", api_key="gsk_...")

api_key is a named field because it is the one argument every pydantic-ai provider accepts. Providers that need more take it verbatim:

Options(
    model="azure:gpt-4o",
    provider_kwargs={
        "azure_endpoint": "https://example.openai.azure.com",
        "api_version": "2024-10-21",
        "api_key": "...",
    },
)

A keyword the named provider does not accept raises TypeError when the provider is constructed, rather than being dropped — a credential silently ignored comes back later as an authentication error that names nothing.

Both apply to every provider inferred from a model string in the run, including fallback_model and compact_model. A run whose models span providers should pass constructed Model instances, which are used as given.

Note that Options.env is unrelated: it is the environment for subprocesses that the Bash tool spawns, and never touches the provider.

Install

uv add ubiquity

The message stream

query() is an async generator. The first message is always a system message describing the resolved configuration; the last is always a result. Tool use, tool results, and assistant turns stream in between.

async for message in query(prompt, options):
    match message.type:
        case "system":       print(message.model, message.tools)
        case "assistant":    print(message.text)
        case "tool_use":     print(message.tool_name, message.tool_input)
        case "tool_result":  print(message.output.content)
        case "result":       print(message.subtype, message.usage)

Built-in tools

Tool	Purpose
Read	Read a file, in cat -n format
Write	Create or overwrite a file
Edit	Exact string replacement
Bash	Run a shell command
Glob	Find files by pattern, newest first
Grep	Search file contents by regex
TodoWrite	Track multi-step work (persistent)
Agent	Delegate to a subagent (added when agents is configured)

Write and Edit enforce read-before-write: an existing file must have been read in full, and must not have changed since, before it can be modified. A partial read (via offset/limit) does not authorize a write, because the writer never saw the part it would discard.

Permissions

Five modes, matching Claude Code:

Mode	Behavior
default	Prompt for anything not pre-approved
acceptEdits	Auto-accept file edits, prompt for the rest
bypassPermissions	Allow everything (deny rules still win)
plan	Read-only; no mutating tool may run
dontAsk	Never prompt; deny anything not pre-approved

Rules are Tool or Tool(matcher), in three forms:

Options(
    allowed_tools=["Bash(git:*)", "Read"],
    disallowed_tools=["Bash(rm:*)"],
    ask_tools=["Bash(git push:*)"],
)

• git:* — prefix; matches git and anything starting git
• git push * — wildcard; * matches any run of characters
• git status — exact

A bare Tool rule also decides availability: disallowed_tools=["Bash"] removes the tool, and setting allowed_tools limits the run to the tools it names. A scoped Tool(matcher) rule never does — Bash(rm:*) leaves Bash exposed and blocks rm at the point of the call.

Two properties are load-bearing and covered by tests:

Deny beats everything, including bypassPermissions. So do user-configured ask rules and safety checks on sensitive paths (.env, .ssh/, .git/).

Allow requires full coverage. A tool may present several candidates for one call — Bash returns each segment of a compound command — and every one must be matched. This is what stops Bash(git:*) from authorizing git status && rm -rf /. Deny and ask fire on any single segment.

To prompt a human, supply can_use_tool:

from ubiquity import PermissionResultAllow, PermissionResultDeny

async def ask_user(tool_name, tool_input, ctx):
    if input(f"Run {tool_name}? [y/N] ").lower() == "y":
        return PermissionResultAllow()
    return PermissionResultDeny(message="User declined.")

Options(can_use_tool=ask_user)

Without a can_use_tool handler, anything that would prompt is denied rather than hanging.

Hooks

Fourteen events, dispatched in registration order. The first hook to block wins and the rest are skipped; a hook that raises is logged and skipped rather than failing the run.

from ubiquity import HookMatcher, HookOutput

async def block_secrets(payload):
    if ".env" in str(payload.tool_input):
        return HookOutput(decision="block", reason="Refusing to touch .env")
    return None

Options(hooks=[HookMatcher("PreToolUse", [block_secrets], matcher="Write|Edit")])

PreToolUse may rewrite the tool input via updated_input; later hooks in the same chain see the rewrite. UserPromptSubmit and SessionStart may inject context via additional_context.

Notification is informational rather than a gate: it fires when a call is waiting on approval and when a run ends by exhausting its turns, raising, or being held open by a Stop hook. payload.extra["reason"] distinguishes them (permission_required, max_turns, error, stopped).

Subagents

A subagent is a nested run with its own history, tool subset, and turn budget. Only its final text returns to the parent, which is the point — the parent's context stays clean.

from ubiquity import AgentDefinition

Options(
    agents={
        "reviewer": AgentDefinition(
            description="Reviews code for correctness",
            prompt="You review diffs and report defects.",
            tools=["Read", "Glob", "Grep"],
            model="anthropic:claude-haiku-4-5-20251001",
        )
    }
)

Isolation is deliberate and partial: a subagent gets fresh file-read bookkeeping, but shares the parent's permission context, because a subagent that could widen its own permissions would be an escalation path. Subagents cannot spawn further subagents, and nesting is capped.

MCP

from ubiquity import parse_config

Options(
    mcp_servers={
        "github": parse_config(
            {
                "command": "npx", 
                "args": [
                    "-y", 
                    "@modelcontextprotocol/server-github"
                ]
            }
        ),
        "docs": parse_config(
            {
                "url": "https://example.test/mcp"
            }
        ),
    }
)

Stdio, SSE, and streamable HTTP are supported. Tools arrive namespaced as mcp__<server>__<tool>, so they cannot shadow a built-in and a whole server can be targeted with mcp__github__*.

Options(disallowed_tools=["mcp__github__*"])

MCP calls go through the same pipeline as a built-in tool — permission rules, PreToolUse and PostToolUse hooks, and tool_use / tool_result messages in the stream. A remote tool is the last thing that should run unobserved.

A server's tools are treated as able to mutate unless it sends a readOnlyHint annotation, so plan mode blocks them by default rather than trusting a server that says nothing about itself.

Compaction

A long run eventually outgrows its context window. Two tiers reclaim it, cheapest first.

Microcompaction costs nothing. The content of older tool results — the file read forty turns ago, the command whose output has long since been acted on — is replaced in place with a marker. No model call, no summary, and the transcript keeps its shape. A microcompact message reports what was cleared.

Only tools whose results are pure observation are eligible (Read, Write, Edit, Bash, Glob, Grep, WebFetch, WebSearch). A tool carrying state the model is expected to still be tracking — TodoWrite, Agent, anything from MCP — is left alone, because clearing it silently rewrites what the model believes about the task.

Full compaction runs only if that leaves the run still over the threshold. The older part of the history is replaced by a model-written summary and the run continues, marked by a compact_boundary message.

Options(
    auto_microcompact=True,
    microcompact_keep_recent=5,
    auto_compact=True,
    max_context_tokens=200_000,
    compact_keep_recent=6,
    compact_model="groq:llama-3.3-70b-versatile",
)

Three details of the second tier are load-bearing.

The cut lands before a model response, never before a request. A request carries the tool results answering the calls in the response above it, so cutting between them would leave results with no matching call — which most providers reject outright. Cutting before a response keeps every pair whole.

The trigger is a token reserve, not a percentage. The threshold is the window minus room for the summary being generated, minus headroom for the next request. A flat 80% would waste 200k tokens on a million-token window and leave a small local model no room to write the summary at all. max_tokens, when set, caps the summary reserve — reserving 20k from a model that can only emit 4k gives back 16k of usable context on every turn.

Pressure is measured past the last usage record. The provider's own accounting is preferred, but the check runs between a response and the request answering it, so the tool results that just landed are never in that count. They are estimated and added on, because a single large file read is precisely the event that pushes a run over the limit. Anything microcompaction just reclaimed is subtracted back out for the same reason in reverse: usage reports what was sent, not what will be sent next.

Repeated failures trip a circuit breaker. A context that is irrecoverably over the limit would otherwise attempt a doomed compaction on every remaining turn; after three consecutive failures the loop stops trying. A failed compaction is never fatal on its own — the history is left alone and the run continues.

There is no built-in table of per-model context windows, because a table asserting sizes for hundreds of models across every provider cannot be kept true, and a stale entry that overstates a window causes exactly the failure compaction exists to prevent. The default is one conservative number; declare the models you actually use:

from ubiquity import register_context_window

register_context_window("gemini-3", 1_048_576)

Options.max_context_tokens overrides per run, and UBIQUITY_MAX_CONTEXT_TOKENS overrides the default globally.

PreCompact can veto a compaction and PostCompact receives the summary.

Todos

TodoWrite edits a list, and the list outlives the run.

Individual tasks can be changed without restating the rest. A task is named by its id or by its exact content, so the model can refer to one either way:

{"add": [{"content": "write the parser"}]}
{"update": [{"task": "write the parser", "status": "in_progress"}]}
{"remove": ["write the parser"]}
{"todos": [...]}

Whole-list writes still work and are the right call when starting a plan from scratch, but they cannot be mixed with edits in one call — a request that both replaces the list and patches it has no unambiguous meaning.

A reference that matches nothing is an error, not a no-op. Ignoring it would leave the model believing it had completed a task it never touched, and a plan that disagrees with reality is worse than a retry.

The one-in-progress invariant is checked against the result, not the request, because with incremental edits an add can introduce a second in-progress task without naming the first.

Lists persist to ~/.ubiquity/todos/<project-slug>/<key>/<task-id>.json, following Claude Code's ~/.claude/tasks directory: one file per task, not one file per list. A list stored as a single document has to be rewritten whole on every change, so two runs editing different tasks from their own stale copies overwrite each other outright. Disjoint tasks in separate files never contend, which removes the problem instead of locking around it. Each write touches only the tasks it changed, and the store is re-read on every call rather than trusting the copy in memory.

The one piece of genuinely shared state is the ordering, carried as a position on each task. Two runs appending at once can pick the same position, which leaves their relative order undefined but loses nothing; ties break on id so a list always reads back stably.

When a stored list has unfinished work, it is loaded into the run and described in the first prompt — a stored list the model is never told about is a list it duplicates. An all-completed list is not carried over, since finished work from an unrelated run is noise.

Options(
    persist_todos=True,
    todo_scope="project",
    todo_dir=None,
)

todo_scope decides what the list belongs to. project keys by working directory, which is what makes a list survive a process exit — every run mints a fresh session id, so a session-scoped list is written and never read again until session resumption is wired up. The tradeoff of the default is real: two concurrent runs in one directory share a list.

Individual task files are written through a temporary file and an atomic rename, so a crash mid-write cannot leave a half-parsed task behind. A file that is unreadable anyway is skipped, not fatal — one corrupt task should not lose the rest of the list.

A subagent gets its own list, keyed by agent id the way Claude Code keys AppState.todos. It shares the parent's working directory, so without a separate key a delegated side task would edit the plan its parent is still working through. The list is discarded when the subagent reports: an agent id names one delegated task and never recurs, so a session spawning hundreds of agents would otherwise accumulate one dead list per agent. Agent ids are unique per invocation because subagents may run in parallel.

Sessions

Transcripts are JSONL, one record per line, appended as the run proceeds — a crashed run still leaves a readable transcript.

from ubiquity import SessionStore

store = SessionStore()
for info in store.list(limit=10):
    print(info.session_id, info.summary)

forked = store.fork(session_id, cwd, up_to_uuid=some_record_uuid)

Records chain through parent_uuid, which is what makes forking work: a fork copies records up to a chosen point and remaps every UUID, producing an independent session that shares history but diverges afterward.

Persistence is on by default under ~/.ubiquity/sessions. Disable with Options(persist_session=False) or redirect with Options(session_dir=...).

Resuming replays a stored transcript as conversation rather than as a summary of one:

Options(resume=session_id)              # continue that session
Options(continue_conversation=True)     # continue the latest one for this cwd
Options(resume=session_id, fork_session=True)   # branch, leaving it untouched

A tool call whose result is missing — denied, or interrupted by a crash — is left out of the replay, since most providers reject a dangling tool use and would make the session unresumable.

Streaming

Options(include_partial_messages=True) adds stream_event messages carrying each delta as it arrives, ahead of the complete assistant message for that turn.

async for message in query(prompt, Options(include_partial_messages=True)):
    if message.type == "stream_event":
        print(message.delta, end="", flush=True)

Settings files

Nothing is read from the filesystem unless asked for, so a stray file cannot reconfigure a caller's run:

Options(setting_sources=["project", "local"])

source	file
user	~/.ubiquity/settings.json
project	<cwd>/.ubiquity/settings.json
local	<cwd>/.ubiquity/settings.local.json

{
  "model": "openai:gpt-5",
  "env": {"NO_COLOR": "1"},
  "permissions": {
    "deny": ["Bash(rm:*)"],
    "ask": ["Bash(git push:*)"],
    "additionalDirectories": ["../shared"]
  }
}

Local beats project beats user, and explicit Options beat all three — except for permission rules, which are unioned. A rule in a settings file is a restriction the caller did not write, so passing a list of their own must not drop it.

Custom tools

Subclass Tool with a Pydantic input model:

from pydantic import BaseModel, Field
from ubiquity import Tool, ToolContext, ToolOutput, PermissionResultAllow, builtin_tools

class SearchInput(BaseModel):
    query: str = Field(description="What to search for.")

class SearchTool(Tool[SearchInput]):
    name = "Search"
    description = "Search the knowledge base."
    input_model = SearchInput

    def is_read_only(self, args): return True
    def is_concurrency_safe(self, args): return True

    async def check_permissions(self, args, ctx):
        return PermissionResultAllow(reason="read-only lookup")

    async def call(self, args, ctx) -> ToolOutput:
        return ToolOutput(content=f"Results for {args.query}")

Options(tools=[*builtin_tools(), SearchTool()])

Override permission_rule_content to make a tool addressable by content rules like Search(internal:*). Return every string that must be authorized — the engine requires all of them to match.

Development

uv sync
uv run pytest

License

MIT