Metadata-Version: 2.4
Name: ubiquity
Version: 0.1.2
Summary: A Claude-Code-style agent SDK built on pydantic-ai, supporting 600+ models across 22 providers
Keywords: agent,sdk,pydantic-ai,llm,claude-code,mcp,tools
Author: Ahmed Saqr
Author-email: Ahmed Saqr <ahmedhassansaqr28@gmail.com>
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Application Frameworks
Classifier: Typing :: Typed
Requires-Dist: anyio>=4.14.2
Requires-Dist: groq>=1.6.0
Requires-Dist: pydantic-ai>=2.16
Requires-Python: >=3.13
Project-URL: Homepage, https://github.com/ahmedhassan456/ubiquity
Project-URL: Repository, https://github.com/ahmedhassan456/ubiquity
Project-URL: Issues, https://github.com/ahmedhassan456/ubiquity/issues
Description-Content-Type: text/markdown

<p align="center">
  <img src="https://raw.githubusercontent.com/ahmedhassan456/ubiquity/main/assets/icon.png" alt="Ubiquity" width="180">
</p>

<h1 align="center">Ubiquity</h1>

<p align="center">
  A Claude-Code-style agent SDK for Python, built on
  <a href="https://ai.pydantic.dev">pydantic-ai</a>.
</p>

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.

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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

```bash
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.

```python
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](#todos)) |
| `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:

```python
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`:

```python
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.

```python
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.

```python
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

```python
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__*`.

```python
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.

```python
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:

```python
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.

## Prompt caching

Providers cache the prefix of a request and re-read it at a fraction of the
normal rate. The cache is theirs — it cannot be inspected, warmed, or addressed
— and the only lever a client has is keeping the prefix identical from one
request to the next.

There is no portable switch, because the providers do not agree on what
caching is. Most cache implicitly with nothing to enable; Anthropic and Bedrock
want an explicit breakpoint; Google wants a separate cached resource created
out of band and billed by the hour. `cache_prompt` is on by default, because an
agent loop resends its whole prefix every turn:

```python
Options(cache_prompt=True)     # 5-minute TTL
Options(cache_prompt="1h")     # longer TTL, higher write cost
Options(cache_prompt=False)    # off
```

| Provider | Mode | Minimum | Read | What `cache_prompt` does |
|---|---|---|---|---|
| Anthropic | explicit | 512–4096 by model | 0.1× | sets `anthropic_cache` |
| Bedrock | explicit | by model | varies | inserts a `CachePoint` |
| OpenAI | automatic | 1024, then 128-token steps | 0.1× | nothing to do |
| Google | implicit | 1024–2048 | ~0.1× | nothing to do |
| Groq | automatic | — | 0.5× | nothing to do |
| DeepSeek | automatic | 64 | ~0.1× | nothing to do |
| xAI | automatic | — | ~0.16× | nothing to do |

The two explicit providers take their breakpoint in different places, which is
why the field is not one setting: Anthropic reads a request-level flag and
advances the breakpoint itself as the conversation grows, while Bedrock takes a
marker inside the message content. Everything else caches on its own, so the
field is deliberately inert there rather than pretending to configure something.

Google's *explicit* caching is not used. It has a 32,768-token minimum and
bills storage by the hour, so an agent loop would pay rent on a cache between
turns; implicit caching already covers recent models for free.

**Conversation history is cached too, not just the system prompt.** On
Anthropic the breakpoint moves forward each request, so turn *n* reads
everything through turn *n−1* from cache and writes only what is new. On
Bedrock the breakpoint sits at the end of the user prompt, which covers the
system prompt, the tool definitions, and the request itself — the part that
does not change while the agent works through its turns.

**A prompt below the minimum is not cached, and no error says so.** Anthropic
needs 512 to 4096 tokens depending on the model, OpenAI 1024, Google 1024 to
2048, DeepSeek 64. A short run reporting zero cache tokens with caching
correctly enabled is under the threshold, not broken.

**Prefix stability is the part that is portable, and it is where the wins
are.** The prefix is the tool definitions, then the system prompt, then the
conversation, and invalidating one level invalidates every level after it — so
an unstable tool description does not cost you the tool block, it costs you the
entire conversation. Two habits keep it intact: nothing time-varying in the
system prompt, and nothing order-varying in the tool definitions. The `Agent`
tool sorts the subagent types it lists for exactly this reason — a set of
agents that renders in a different order between runs costs a full miss for a
difference no model can see.

**Breaks are silent.** Nothing fails and nothing warns; the only symptom is a
bill several times larger than it should be. `detect_cache_breaks` watches the
cache-read count reported with each response, and when it falls by more than
5% *and* 2000 tokens, names whichever input changed:

```python
Options(detect_cache_breaks=True)
```

```
prompt cache break: tool schema changed (Agent) [call #7, cache read 48210 -> 0, written 48355]
```

Both thresholds have to clear: a proportional test alone fires constantly on
small conversations, an absolute one alone misses large ones. Compaction resets
the baseline, since it drops history on purpose and reporting that as a break
is how warnings become noise. Tools are hashed individually as well as
together, so a rewritten description is named even when the tool set is
unchanged — upstream found that to be the majority of tool-caused breaks and
the hardest to spot by reading code.

Detection is off by default and costs a few hashes per request when on. It
covers the main run, not subagents, which do not expose per-request usage.
Providers that report no cache tokens hold the count at zero, so they stay
silent rather than reporting nonsense — no signal is not the same as no cache.

## 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:

```python
{"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.

```python
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.

```python
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:

```python
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.

```python
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:

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

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

```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:

```python
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

```bash
uv sync
uv run pytest
```

## License

MIT
