jeevesagent.core.protocols

Protocol definitions for every module boundary.

These structural types are the contract surface of the harness. Every implementation — first-party or third-party — satisfies one of these. The loop and the agent only depend on the protocols, never on concrete implementations.

The protocols are intentionally async-only: every method that performs I/O is a coroutine, every stream is an AsyncIterator, every resource is an AsyncContextManager.

Classes

Budget	Resource governance — tokens, calls, cost, wall clock.
Embedder	Text-to-vector embedding model used by the memory subsystem.
HookHost	Aggregator over user-registered lifecycle callbacks.
Memory	Tiered memory: working blocks, episodic store, semantic graph.
Model	LLM provider interface. One adapter per lab (Anthropic, OpenAI, ...).
Permissions	Decides whether a tool call is allowed.
Runtime	Durable execution. Wraps every side effect in a journal entry.
RuntimeSession	Handle to an open durable session held by a Runtime.
Sandbox	Isolation layer for tool execution.
Secrets	Resolution and redaction of named secrets.
Telemetry	OpenTelemetry-compatible tracing/metrics surface.
ToolHost	MCP-aware tool registry. Lazy-loads schemas on demand.

Module Contents

class jeevesagent.core.protocols.Budget

Bases: Protocol

Resource governance — tokens, calls, cost, wall clock.

async allows_step() → jeevesagent.core.types.BudgetStatus

async consume(*, tokens_in: int, tokens_out: int, cost_usd: float) → None

class jeevesagent.core.protocols.Embedder

Bases: Protocol

Text-to-vector embedding model used by the memory subsystem.

async embed(text: str) → list[float]

async embed_batch(texts: list[str]) → list[list[float]]

dimensions: int

name: str

class jeevesagent.core.protocols.HookHost

Bases: Protocol

Aggregator over user-registered lifecycle callbacks.

async on_event(event: jeevesagent.core.types.Event) → None

async post_tool(call: jeevesagent.core.types.ToolCall, result: jeevesagent.core.types.ToolResult) → None

async pre_tool(call: jeevesagent.core.types.ToolCall) → jeevesagent.core.types.PermissionDecision

class jeevesagent.core.protocols.Memory

Bases: Protocol

Tiered memory: working blocks, episodic store, semantic graph.

async append_block(name: str, content: str) → None

Append to a named block, creating it if absent.

async consolidate() → None

Background: extract semantic facts from recent episodes.

async recall(query: str, *, kind: str = 'episodic', limit: int = 5, time_range: tuple[datetime.datetime, datetime.datetime] | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Episode]

Retrieve episodes (or facts, when kind='semantic').

When user_id is supplied, results are restricted to episodes stored with that exact user_id value. None is its own bucket (the “anonymous / single-tenant” namespace) — episodes stored with user_id=None are never visible to a query with user_id="alice" and vice versa. Backends MUST honour this filter to preserve the framework’s multi-tenant safety contract.

async recall_facts(query: str, *, limit: int = 5, valid_at: datetime.datetime | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

Retrieve bi-temporal facts matching query.

Backends that don’t expose a fact store return []. The agent loop calls this directly rather than duck-typing on memory.facts so backends without fact support don’t need any opt-out mechanism.

user_id filters by namespace partition with the same semantics as recall(): None is its own bucket and does not cross-contaminate with non-None values.

async remember(episode: jeevesagent.core.types.Episode) → str

Persist an episode. Returns the episode ID.

async session_messages(session_id: str, *, user_id: str | None = None, limit: int = 20) → list[jeevesagent.core.types.Message]

Return the most-recent limit user/assistant turns from the conversation identified by session_id, in order (oldest first).

This is the conversation-continuity primitive — the agent loop calls it at the top of every run so that reusing a session_id actually continues the chat (the model sees previous turns as real Message history) rather than starting fresh and relying solely on semantic recall.

user_id MUST be respected by backends as a hard namespace partition: messages persisted under one user_id are never visible to a query scoped to a different one. Backends without persisted message logs return [] — the agent loop falls back to the semantic-recall path in that case.

async update_block(name: str, content: str) → None

Replace the contents of a named block.

async working() → list[jeevesagent.core.types.MemoryBlock]

All in-context blocks. Pinned to every prompt.

class jeevesagent.core.protocols.Model

Bases: Protocol

LLM provider interface. One adapter per lab (Anthropic, OpenAI, …).

The required surface is stream(...) — every adapter must implement it. Adapters MAY additionally override complete(...) with a non-streaming (single-shot) call; if not, complete falls back to consuming the stream internally and assembling the full response, which is correct but slower (per-chunk wire + parsing overhead). Architectures use complete on the non-streaming hot path (agent.run()) and stream when a consumer is reading from agent.stream().

stream(messages: list[jeevesagent.core.types.Message], *, tools: list[jeevesagent.core.types.ToolDef] | None = None, temperature: float = 1.0, max_tokens: int | None = None) → collections.abc.AsyncIterator[jeevesagent.core.types.ModelChunk]

Stream completion chunks. Each chunk is text, tool_call, or finish.

name: str

class jeevesagent.core.protocols.Permissions

Bases: Protocol

Decides whether a tool call is allowed.

async check(call: jeevesagent.core.types.ToolCall, *, context: collections.abc.Mapping[str, Any]) → jeevesagent.core.types.PermissionDecision

class jeevesagent.core.protocols.Runtime

Bases: Protocol

Durable execution. Wraps every side effect in a journal entry.

session(session_id: str) → contextlib.AbstractAsyncContextManager[RuntimeSession]

Open or resume a durable session.

async signal(session_id: str, name: str, payload: Any) → None

Send an external signal (e.g., human approval) to a session.

async step(name: str, fn: collections.abc.Callable[Ellipsis, collections.abc.Awaitable[Any]], *args: Any, idempotency_key: str | None = None, **kwargs: Any) → Any

Execute fn as a journaled step. Replays cached on resume.

stream_step(name: str, fn: collections.abc.Callable[Ellipsis, collections.abc.AsyncIterator[Any]], *args: Any, **kwargs: Any) → collections.abc.AsyncIterator[Any]

Execute a streaming step. Replays the aggregate on resume.

name: str

class jeevesagent.core.protocols.RuntimeSession

Bases: Protocol

Handle to an open durable session held by a Runtime.

async deliver(name: str, payload: Any) → None

id: str

class jeevesagent.core.protocols.Sandbox

Bases: Protocol

Isolation layer for tool execution.

async execute(tool: jeevesagent.core.types.ToolDef, args: collections.abc.Mapping[str, Any]) → jeevesagent.core.types.ToolResult

with_filesystem(root: str) → contextlib.AbstractAsyncContextManager[None]

Temporary filesystem sandbox for the duration of the context.

class jeevesagent.core.protocols.Secrets

Bases: Protocol

Resolution and redaction of named secrets.

redact(text: str) → str

async resolve(ref: str) → str

async store(ref: str, value: str) → None

class jeevesagent.core.protocols.Telemetry

Bases: Protocol

OpenTelemetry-compatible tracing/metrics surface.

async emit_metric(name: str, value: float, **attrs: Any) → None

trace(name: str, **attrs: Any) → contextlib.AbstractAsyncContextManager[jeevesagent.core.types.Span]

class jeevesagent.core.protocols.ToolHost

Bases: Protocol

MCP-aware tool registry. Lazy-loads schemas on demand.

async call(tool: str, args: collections.abc.Mapping[str, Any], *, call_id: str = '') → jeevesagent.core.types.ToolResult

Invoke tool with args. The call_id is propagated into the returned ToolResult so the loop can correlate results with the originating model-emitted call.

async list_tools(*, query: str | None = None) → list[jeevesagent.core.types.ToolDef]

watch() → collections.abc.AsyncIterator[jeevesagent.core.types.ToolEvent]

Notifications when the tool list changes (MCP listChanged).