jeevesagent.core.context¶

Per-run context propagation.

A single RunContext is built at the top of every Agent.run() (or Agent.stream()) call and stored in a contextvars.ContextVar for the duration of the run. Tools, hooks, sub-agents, and architectures all read it through get_run_context() rather than threading it through every signature.

The framework treats user_id and session_id as first-class typed primitives — not strings buried in a free-form configurable dict. user_id partitions memory recall; session_id identifies the conversation thread for replay and continuity. Application- specific keys go in the metadata mapping, where the framework makes no claim to understand them.

The contextvar is automatically propagated by anyio’s structured concurrency primitives (create_task_group, start_soon), so parallel tool dispatch, sub-agent spawning, and streaming consumers all see the same context without any explicit plumbing.

Tests that call @tool functions directly (no active agent run) get a default empty RunContext rather than an exception — preserving direct-invocation ergonomics.

Exceptions¶

IsolationWarning

Emitted when a memory query is likely to silently miss data

Classes¶

`RunContext`	Typed, immutable context for one agent run.
`set_run_context`	Context manager that installs a `RunContext` for the

Functions¶

get_run_context(→ RunContext)

Return the RunContext for the currently-running agent.

Module Contents¶

exception jeevesagent.core.context.IsolationWarning[source]¶

Bases: UserWarning

Emitted when a memory query is likely to silently miss data because the caller forgot to pass user_id.

Concrete trigger: a backend’s recall / recall_facts runs with user_id=None against a store whose persisted records include at least one non-None user_id — the partition is safe (the anonymous bucket and named-user buckets are isolated), but the developer probably wired up multi-tenancy somewhere and forgot to pass user_id here, so they will see suspiciously empty recall results.

Subclass of UserWarning so it goes through Python’s standard warnings filter machinery — apps can silence, promote-to-error, or log it however they want, e.g.:

import warnings
from jeevesagent import IsolationWarning
warnings.simplefilter("error", IsolationWarning)  # raise on hit

Initialize self. See help(type(self)) for accurate signature.

class jeevesagent.core.context.RunContext[source]¶

Typed, immutable context for one agent run.

Set once at the start of Agent.run() and propagated to every architecture, tool, hook, sub-agent, and memory operation via a contextvars.ContextVar. The framework treats user_id and session_id as first-class fields (typed, namespaced); metadata is an opaque bag for app-specific keys the framework does not interpret.

Construct one directly when you need to spawn work outside an active run with explicit scope:

ctx = RunContext(user_id="alice", session_id="conv_42")
async with set_run_context(ctx):
    await my_tool(...)

Inside an agent run, prefer get_run_context() over constructing a new one — that gives you the live context the framework set up.

get(key: str, default: Any = None) → Any[source]¶: Shorthand for self.metadata.get(key, default).

Return a new context with selected fields replaced.

Used by multi-agent architectures when spawning sub-agents that need to inherit most of the parent’s context but with a derived session_id or augmented metadata. The sentinel makes “leave this field unchanged” distinguishable from “explicitly set this field to None”.

metadata: collections.abc.Mapping[str, Any]¶: Free-form application context. Use this for keys the framework does not need to understand — locale, request id, feature flags, tenant id beyond user_id, etc. Read inside tools / hooks via get_run_context().metadata.

run_id: str = ''¶: Unique identifier for this single Agent.run() invocation. Distinct from session_id (which identifies a conversation that may span many runs). Auto-set by Agent.run(); an explicit value passed in by the caller is overridden.

session_id: str | None = None¶: Conversation thread identifier. Reusing the same session_id across calls signals “continue this conversation” — the framework will rehydrate prior session messages so the model sees real chat history, not just memory recall. None means “fresh conversation”; the framework auto-generates one inside Agent.run() if not supplied.

user_id: str | None = None¶: Namespace for memory recall + persistence. None is the “anonymous / single-tenant” bucket; episodes / facts stored with user_id=None never see episodes / facts stored with a non-None user_id and vice versa. The framework treats this as a hard partition key, not a soft filter.

class jeevesagent.core.context.set_run_context(context: RunContext)[source]¶

Context manager that installs a RunContext for the duration of an async with block.

The framework uses this internally inside Agent.run() to expose the live context to tools and hooks. Application code rarely needs it, but it is the supported way to invoke a tool outside an agent loop with explicit scope — for example in background workers that share tool implementations with the agent:

async with set_run_context(RunContext(user_id="alice")):
    await some_tool(...)

Behaves correctly under structured concurrency: nested async with blocks restore the prior context on exit, and anyio task-group spawns inherit the active context automatically.

jeevesagent.core.context.get_run_context() → RunContext[source]¶

Return the RunContext for the currently-running agent.

Inside an active Agent.run() call this returns the live context with user_id, session_id, run_id, and metadata populated. Outside any active run (test code, direct @tool invocation, REPL exploration) this returns the default empty RunContext — never raises.

Tools that need scope information call this rather than taking extra parameters:

@tool
async def fetch_user_orders() -> str:
    ctx = get_run_context()
    return await db.query("orders", user_id=ctx.user_id)