Metadata-Version: 2.4
Name: ubiquity
Version: 0.1.1
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.

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