API Reference

API documentation will be added as implementation progresses.

Current Modules

`yoker.api`

Thin Pythonic facade (MBI-003) over the real Agent and Session classes. Exposes the one-shot helpers process / do, the reusable-agent builder agent, the multi-turn session async context manager, and the run_sync sync wrapper. yoker.agent is the factory function re-exported from this module (not a module itself).

Thin Pythonic API for Yoker (MBI-003).

A minimal facade over the real yoker.session.Session and yoker.core.Agent classes:

process() — one-shot turn; builds an agent, runs a single prompt, returns the response string. The agent is discarded.

do() — one-shot skill invocation; builds an agent, runs Agent.do(), returns the response string.

agent() — builder that returns a configured, reusable yoker.Agent.

session() — async context manager yielding the real yoker.session.Session with a registered primary agent.

run_sync() wraps asyncio.run() for synchronous callers (scripts, notebooks, REPLs). It is the only sync entry point — there are no per-call *_sync variants.

Build a configured, reusable Agent.

Convenience factory over the Agent constructor. Maps the most common configuration knobs (model, provider, system prompt, tools, skills, plugins, thinking mode, event handler) to keyword arguments so programmatic callers do not need to build a Config or AgentDefinition by hand.

The returned agent is reusable across turns and async tasks. For a stateless one-shot call see yoker.process(); for multi-turn conversations with context persistence see yoker.session().

Parameters:

model – Optional model override applied to the active provider config.
provider – Optional backend provider ("ollama", "openai", …).
system_prompt – Optional override for the agent’s system prompt.
tools – Whitelist of tool names, the ALL_TOOLS sentinel, or None. The default (ALL_TOOLS) grants all config-enabled tools; omit the arg (or pass ALL_TOOLS explicitly) for all tools. None disables all tools (matches the AgentDefinition contract); [] also disables all; ["read"] keeps only read.
skills – Optional whitelist of skill names. None keeps all loaded skills; [] disables all; ["commit"] keeps only commit.
plugins – Optional plugin packages to load (e.g. ["pkgq"]).
agent_path – Optional path to an agent definition Markdown file.
agent_definition – Optional explicit AgentDefinition. Takes precedence over agent_path.
thinking – Thinking mode for the model.
event_handler – Optional callback (sync or async) receiving every event emitted by the agent.
config – Optional explicit Config. When omitted and no model / provider / plugins overrides are given, the Agent constructor discovers the config from the filesystem via get_yoker_config(). When provided (or when overrides need applying), a programmatic Config is used as the base.
context_manager – Optional ContextManager for the agent. When omitted a SimpleContextManager is used (in-memory, no persistence).
backend – Optional ModelBackend (e.g. shared from a yoker.session.Session). When provided the agent reuses it instead of creating one from config.
console_logging – Whether to enable console logging. The one-shot helpers (process(), do()) and session() set this to False so they stay quiet; the default here matches the Agent constructor.

Returns:

A fully constructed yoker.Agent instance.

async yoker.api.do(skill_name: str, prompt: str, args: str = '', **common_kwargs: Any) → str[source]

Invoke a skill as a one-shot command and return the response.

Builds an agent via yoker.agent(), runs Agent.do() with the given skill, and returns the assistant’s reply. The agent is discarded (stateless one-shot).

Parameters:

skill_name – Name of the skill to invoke (bare or namespaced).
prompt – The user’s task. Sent as the user message after the skill context is injected.
args – Optional arguments forwarded to the skill’s invocation block.
**common_kwargs – Same builder kwargs as yoker.agent().

Returns:

The assistant’s response string for the turn.

async yoker.api.process(prompt: str, **common_kwargs: Any) → str[source]

Ask the configured model a single question and return the response.

Loads a Config, constructs an Agent, runs one turn, and returns the assistant’s reply. The agent is discarded (stateless one-shot). For multi-turn conversations use yoker.session(); for a reusable agent use yoker.agent().

Parameters:

prompt – The user’s question or instruction.
**common_kwargs – Same builder kwargs as yoker.agent() (model, provider, system_prompt, tools, skills, plugins, thinking, event_handler, config).

Returns:

The assistant’s response string for the turn.

yoker.api.run_sync(coro: Coroutine[Any, Any, _T]) → _T[source]

Run a coroutine from synchronous code.

Uses asyncio.run() so a fresh event loop is created and torn down per call — appropriate for the sync convenience path, which by definition is not latency-sensitive. Raises a clear error when called from inside a running loop so users reach for the async variant instead.

Parameters:: coro – The coroutine to run to completion.
Returns:: The coroutine’s result.
Raises:: RuntimeError – If an event loop is already running in the current thread.

Open a multi-turn session with automatic context persistence.

Builds on the real yoker.session.Session (MBI-007). The primary Agent is constructed inside Session.__init__() via the unified _create_agent flow with the given builder kwargs. The primary agent is available via Session.agent; sub-agents can be spawned via await session.spawn(name). Event handlers are registered via session.on_event(...).

Parameters:

id – Optional session id. When the id matches an existing persisted session the conversation history is resumed. When omitted a fresh UUID-based id is generated.
persist – When True (default) the primary agent’s context is wrapped with yoker.context.Persisted so turns survive across sessions with the same id. Set to False for an in-memory session.
fresh – When True, ignore any persisted state for the given id and start with an empty context.
model – Optional model override applied to the active provider config.
provider – Optional backend provider ("ollama", "openai", …).
system_prompt – Optional override for the agent’s system prompt.
tools – Whitelist of tool names, the ALL_TOOLS sentinel, or None. The default (ALL_TOOLS) grants all config-enabled tools; omit the arg (or pass ALL_TOOLS explicitly) for all tools. None disables all tools (matches the AgentDefinition contract); [] also disables all; ["read"] keeps only read.
skills – Optional whitelist of skill names. None keeps all loaded skills; [] disables all; ["commit"] keeps only commit.
plugins – Optional plugin packages to load (e.g. ["pkgq"]).
agent_path – Optional path to an agent definition Markdown file.
agent_definition – Optional explicit AgentDefinition. Takes precedence over agent_path.
thinking – Thinking mode for the model.
event_handler – Optional callback (sync or async) receiving every event emitted by the agent.
config – Optional explicit Config. When omitted and no model / provider / plugins overrides are given, the Agent constructor discovers the config from the filesystem via get_yoker_config(). When provided (or when overrides need applying), a programmatic Config is used as the base.

Yields:

The real yoker.session.Session with a registered primary agent.

`yoker.core`

Agent layer (no UI, no Session coupling). Provides the unified Agent class: async-only, emits Event objects, and exposes on_event() (the single event-handler registration method), process(), and do().

Asynchronous Agent implementation for Yoker.

class yoker.core.Agent(config: Config | None = None, thinking_mode: ThinkingMode = ThinkingMode.ON, agent_definition: AgentDefinition | None = None, agent_path: Path | str | None = None, context_manager: ContextManager | None = None, plugins: tuple[str, ...] = (), backend: ModelBackend | None = None, parse_cli_args: bool = False, console_logging: bool = True)[source]

Bases: object

Asynchronous agent that chats with model backends and uses tools.

async do(skill_name: str, prompt: str, args: str = '') → str[source]

Invoke a skill as a command on this agent and return the response.

Loads the skill’s context into the conversation (via inject_skill_context()) and then runs a single process() turn. The skill must be discoverable in the agent’s skill registry (loaded from configured directories or plugins).

Parameters:

skill_name – Name of the skill to invoke (bare or namespaced).
prompt – The user’s task. Sent as the user message after the skill context is injected. May be empty when the skill content alone is enough to drive the turn.
args – Optional arguments forwarded to the skill’s invocation block.

Returns:

The assistant’s response string for the turn.

static filter_skills(registry: SkillRegistry, requested: list[str]) → None[source]

Keep only requested skills in registry; raise on unknown names.

Mutates registry in place. Bare names try the yoker: namespace as a fallback; namespaced names must match exactly.

property guardrail: PathGuardrail

Return the path guardrail for file system operations.

Returns:: The guardrail instance for path validation.
Return type:: PathGuardrail

inject_skill_context(skill_name: str, args: str | None = None) → None[source]: Inject skill context into the conversation.

Register an event handler and return it for chaining.

Accepts a sync or async callable accepting an yoker.events.Event (or a yoker.events.SessionEvent envelope when the agent is part of a yoker.session.Session).

Parameters:: handler – A callable accepting an yoker.events.Event. May be sync or async.
Returns:: The same handler callable, so callers can chain or inline the registration (e.g. agent.on_event(print)).

async process(message: str) → str[source]

Process a single message and return the response.

Concurrent process() calls on the same agent are serialized via an internal asyncio.Queue. When a turn is in flight, additional calls wait in the queue and are processed strictly one at a time — the backend never sees parallel chat_stream invocations on the same agent. The public API is unchanged: callers simply await agent.process(msg) and the queueing is transparent.

Cancels the consumer task on cancellation, propagating CancelledError to the awaiting caller.

`yoker.session`

Multi-turn session construct (MBI-007): an async context manager owning a team of agents. The primary agent is available via Session.agent; sub-agents can be spawned via Session.spawn(). Inter-agent messaging uses Session.send(*, to, from_, content) with plain strings. Event handlers are registered via Session.on_event(...).

Session class — async context manager owning a team of agents.

A Session is the container+coordinator for a team of agents. It owns:

the session id namespace
the name→agent map
the yoker.agents.AgentRegistry
event aggregation handlers
shared backends

class yoker.session.Session(config: Config, *, session_id: str | None = None, extra_plugins: tuple[str, ...] = (), agent_definition: AgentDefinition | None = None, agent_path: str | Path | None = None, plugins: tuple[str, ...] | None = None, thinking_mode: ThinkingMode | None = None, console_logging: bool = False)[source]

Bases: object

Async context manager that owns a team of agents.

Parameters:

config – The root Config the session reads session-level settings from (config.session and config.tools.agent).
session_id – Optional explicit session id. When omitted a UUID-based id is generated.

get_agent(name: str) → Agent | None[source]

Look up an active agent by its session-assigned name.

Returns None when no active agent has the given name (including when the agent has finished and been removed.

get_backend(config: Config) → ModelBackend[source]

Return a shared or fresh backend for the given config.

Backends are cached by provider config signature. Agents that share the same active provider config reuse the same backend instance. Per-agent model/provider overrides produce a different signature and therefore a fresh backend.

inject_tools(agent: Agent, agent_id: str) → None[source]

Inject Session-injected tools onto an agent

Registers agent and send_message on the agent’s tool registry. The Session captures itself in the closure (back-reference) so the tools can call session.spawn / session.send at execution time. ListAgents is deferred and is NOT injected.

agent is gated by config.tools.agent.enabled (the existing global kill-switch). send_message is always injected when an agent is part of a session.

Parameters:

agent – The Agent to inject tools onto.
agent_id – The agent’s session-assigned runtime id (used as Message.from_id by send_message).

Register a session-scoped event handler and return it for chaining.

Handlers receive events emitted by the Session (session-level events) and agent events wrapped in a SessionEvent envelope.

release(agent: Agent) → None[source]

Release a spawned agent: emit AGENT_FINISHED and remove from the active map.

Removes the agent by identity. When the agent is not registered (already released or never registered) this is a no-op. This is the single cleanup path used by both the agent tool and standalone callers that drive a spawned agent directly.

async send(*, to: Agent, from_: Agent, content: str) → str[source]

Send a message from one agent to another and return the target’s reply.

The Python API accepts Agent instances directly. The session-assigned ids are resolved via reverse-lookup against the active map and used only for the AgentMessageEvent payload. Emits the event, then calls await to.process(content). Request-response only — content is a plain string and the response is a plain string (no streaming).

Parameters:

to – The target Agent (must be active in this session).
from – The sending Agent.
content – Plain-string message content (the prompt).

Returns:

The target agent’s response string. When the target agent raises, the exception is caught and an error string is returned (preserving the agent tool’s behaviour of not propagating exceptions).

Raises:

ValueError – When the target agent is not registered in this session.

async spawn(name: str, *, requester: Agent | None = None) → Agent[source]

Spawn a child agent by name and return it (no prompt is run).

The canonical sub-agent API (MBI-003 Decision 8). Thin wrapper over _spawn_internal() that returns only the constructed Agent. The caller drives the agent directly (e.g. await agent.process("...")). The agent stays registered in the session’s active map until release() is called (or the session exits and cleans up).

Parameters:

name – Agent definition name (bare or namespaced) to spawn.
requester – The requesting Agent (for allowlist enforcement). When None (top-level spawn) the allowlist check is bypassed.

Returns:

The spawned Agent instance.

Raises:

ValueError – On allowlist violation, unknown agent, or capacity (max_agents) exceeded.

`yoker.events`

Event system for library-first design. The Agent emits events that handlers can subscribe to.

Event system for Yoker agents.

class yoker.events.AgentFinishedEvent(session_id: str, agent_id: str, *, type: EventType, timestamp: datetime = <factory>)[source]

Bases: Event

Emitted when an agent finishes in a Session.

This is a lifecycle signal; the agent is removed from the Session’s active list after this event is emitted.

session_id

The session that owned the agent.

Type:: str

agent_id

The unique session-assigned id of the finished agent.

Type:: str

agent_id: str

session_id: str

class yoker.events.AgentMessageEvent(session_id: str, from_id: str, to_id: str, content: str, *, type: EventType, timestamp: datetime = <factory>)[source]

Bases: Event

Emitted when an inter-agent message is routed through a Session.

session_id

The session routing the message.

Type:: str

from_id

The unique id of the sending agent.

Type:: str

to_id

The unique id of the receiving agent.

Type:: str

content

The plain-string message content.

Type:: str

content: str

from_id: str

session_id: str

to_id: str

class yoker.events.AgentSpawnedEvent(session_id: str, agent_id: str, definition_name: str, *, type: EventType, timestamp: datetime = <factory>)[source]

Bases: Event

Emitted when an agent is spawned into a Session.

session_id

The session that owns the spawned agent.

Type:: str

agent_id

The unique session-assigned id of the spawned agent.

Type:: str

definition_name

The agent definition name the agent was created from.

Type:: str

agent_id: str

definition_name: str

session_id: str

class yoker.events.CommandEvent(command: str, result: str, *, type: EventType, timestamp: datetime = <factory>)[source]

Bases: Event

Emitted when a slash command is executed.

command: str

result: str

class yoker.events.ContentChunkEvent(text: str, content_type: str = 'text/plain', *, type: EventType, timestamp: datetime = <factory>)[source]

Bases: Event

Emitted for each chunk of content output.

text

The content text chunk.

Type:: str

content_type

MIME type of the content (default: “text/plain”). Possible values: “text/plain”, “text/markdown”, “text/html”, etc.

Type:: str

content_type: str = 'text/plain'

text: str

class yoker.events.ContentEndEvent(total_length: int, *, type: EventType, timestamp: datetime = <factory>)[source]

Bases: Event

Emitted when content output ends.

total_length: int

class yoker.events.ContentStartEvent(*, type: EventType, timestamp: datetime = <factory>)[source]

Bases: Event

Emitted when content output begins.

class yoker.events.Event(*, type: EventType, timestamp: datetime = <factory>)[source]

Bases: object

Base event class with common fields.

timestamp: datetime

type: EventType

class yoker.events.EventRecorder(path: Path)[source]

Bases: object

Records all events to a JSONL file for replay.

This class is an event handler that can be registered with an Agent to capture all events to a JSONL file. The file can later be replayed using EventReplayAgent.

Example

agent = Agent(config=config) recorder = EventRecorder(Path(“session.jsonl”)) agent.on_event(recorder) # … run session … recorder.close()

close() → None[source]: Close the recording file.

class yoker.events.EventType(*values)[source]

Bases: Enum

Enumeration of all event types.

AGENT_FINISHED = 16

AGENT_MESSAGE = 17

AGENT_SPAWNED = 15

COMMAND = 12

CONTENT_CHUNK = 7

CONTENT_END = 8

CONTENT_START = 6

SESSION_END = 14

SESSION_START = 13

THINKING_CHUNK = 4

THINKING_END = 5

THINKING_START = 3

TOOL_CALL = 9

TOOL_CONTENT = 11

TOOL_RESULT = 10

TURN_END = 2

TURN_START = 1

class yoker.events.SessionEndEvent(session_id: str, *, type: EventType, timestamp: datetime = <factory>)[source]

Bases: Event

Emitted when a Session ends.

session_id

The unique session identifier.

Type:: str

session_id: str

class yoker.events.SessionEvent(agent_id: str, event: Event)[source]

Bases: object

Envelope wrapping an Event with its source agent’s id.

agent_id

The unique session-assigned id of the agent that produced the wrapped event.

Type:: str

event

The original event, unchanged.

Type:: yoker.events.types.Event

agent_id: str

event: Event

class yoker.events.SessionStartEvent(session_id: str, *, type: EventType, timestamp: datetime = <factory>)[source]

Bases: Event

Emitted when a Session starts.

session_id

The unique session identifier.

Type:: str

session_id: str

class yoker.events.ThinkingChunkEvent(text: str, *, type: EventType, timestamp: datetime = <factory>)[source]

Bases: Event

Emitted for each chunk of thinking output.

text: str

class yoker.events.ThinkingEndEvent(total_length: int, *, type: EventType, timestamp: datetime = <factory>)[source]

Bases: Event

Emitted when thinking output ends.

total_length: int

class yoker.events.ThinkingStartEvent(*, type: EventType, timestamp: datetime = <factory>)[source]

Bases: Event

Emitted when thinking output begins.

class yoker.events.ToolCallEvent(tool_name: str, arguments: dict[str, ~typing.Any], *, type: ~yoker.events.types.EventType, timestamp: ~datetime.datetime = <factory>)[source]

Bases: Event

Emitted when a tool is called.

arguments: dict[str, Any]

tool_name: str

class yoker.events.ToolContentEvent(tool_name: str, operation: str, path: str, content_type: str, content: str | None = None, metadata: dict[str, ~typing.Any]=<factory>, *, type: EventType, timestamp: datetime = <factory>)[source]

Bases: Event

Emitted when a tool has content to display (write/update operations).

tool_name

Name of the tool (e.g., “write”, “update”).

Type:: str

operation

Operation type (e.g., “write”, “replace”, “insert_before”, “insert_after”, “delete”).

Type:: str

path

Resolved file path.

Type:: str

content_type

MIME type of the content. Common values: - “text/plain”: Plain text content (default) - “text/x-diff”: Unified diff format - “application/json”: JSON data - “text/markdown”: Markdown content - “application/x-summary”: Custom type indicating operation summary only (no content field)

Type:: str

content

Content to display (truncated if too large, None for summary type).

Type:: str | None

metadata

Additional metadata (lines, bytes, is_new_file, is_overwrite, etc.).

Type:: dict[str, Any]

content: str | None = None

content_type: str

metadata: dict[str, Any]

operation: str

path: str

tool_name: str

class yoker.events.ToolResultEvent(tool_name: str, result: str, success: bool = True, *, type: EventType, timestamp: datetime = <factory>)[source]

Bases: Event

Emitted when a tool returns a result.

result: str

success: bool = True

tool_name: str

class yoker.events.TurnEndEvent(response: str, tool_calls_count: int = 0, input_tokens: int = 0, output_tokens: int = 0, prompt_eval_count: int = 0, eval_count: int = 0, total_duration_ms: int = 0, *, type: EventType, timestamp: datetime = <factory>)[source]

Bases: Event

Emitted when processing a user message completes.

eval_count: int = 0

input_tokens: int = 0

output_tokens: int = 0

prompt_eval_count: int = 0

response: str

tool_calls_count: int = 0

total_duration_ms: int = 0

class yoker.events.TurnStartEvent(message: str, *, type: EventType, timestamp: datetime = <factory>)[source]

Bases: Event

Emitted when processing a user message begins.

message: str

yoker.events.deserialize_event(entry: dict[str, Any]) → Event | SessionEvent[source]

Deserialize a dictionary back to an event object.

Looks up the event class via EVENT_CLASS_MAP and constructs it directly from the stored data plus the envelope fields, avoiding per-type dispatch.

When the entry carries the session_event marker, the SessionEvent envelope is reconstructed around the deserialized inner event.

Parameters:

entry – Dictionary with type, timestamp, and data keys, or the session_event envelope form.

Returns:

Reconstructed event object (bare Event or SessionEvent).

Raises:

KeyError – If required fields are missing.
ValueError – If event type is unknown.

yoker.events.serialize_event(event: Event | SessionEvent) → dict[str, Any][source]

Serialize an event to a JSON-serializable dictionary.

Uses dataclasses.asdict() to convert the event dataclass into a dict, then strips the type and timestamp envelope fields (which are returned separately as part of the wrapper). The remaining dict is the event’s data payload.

When the input is a SessionEvent envelope, the wrapper is serialized alongside the inner event:

{
  "session_event": True,
  "agent_id": "<source agent id>",
  "event": <serialize_event(inner)>,
}

Parameters:: event – The event to serialize (bare Event or SessionEvent).
Returns:: Dictionary with type, timestamp, and data keys (or the session_event envelope form for SessionEvent inputs).

Event Types:

Event	Description
`TurnStartEvent`	Emitted when processing a user message begins
`TurnEndEvent`	Emitted when processing a user message completes
`ThinkingStartEvent`	Emitted when LLM thinking output begins
`ThinkingChunkEvent`	Emitted for each chunk of thinking output
`ThinkingEndEvent`	Emitted when thinking output ends
`ContentStartEvent`	Emitted when content output begins
`ContentChunkEvent`	Emitted for each chunk of content output
`ContentEndEvent`	Emitted when content output ends
`ToolCallEvent`	Emitted when a tool is called
`ToolResultEvent`	Emitted when a tool returns a result
`ToolContentEvent`	Emitted when a tool produces display content
`CommandEvent`	Emitted when a slash command is replayed

Handlers are plain callables that receive Event objects. Register them with agent.on_event(...) (or session.on_event(...) for session-scoped handlers).

`yoker.ui`

User interface layer. Provides UIHandler implementations and the UIBridge that routes agent events to UI methods.

`yoker.tools`

Tools framework for Yoker.

Provides the tool framework including result types, guardrails, registry, annotation markers, and context. Built-in tools are in yoker.builtin.

class yoker.tools.FetchedContent(url: str, title: str, content: str, content_type: str = 'markdown', source: str = 'unknown', metadata: dict[str, ~typing.Any]=<factory>)[source]

Bases: object

Content fetched from a web URL.

url

The URL that was fetched.

Type:: str

title

Page title (extracted or derived).

Type:: str

content

Fetched content (markdown, text, or original).

Type:: str

content_type

Content format (“markdown”, “text”, “html”).

Type:: str

source

Backend that fetched this content (“ollama”, “local”).

Type:: str

metadata

Additional metadata (e.g., links, images, word_count).

Type:: dict[str, Any]

content: str

content_type: str = 'markdown'

classmethod from_dict(data: dict[str, Any]) → FetchedContent[source]

Create from dictionary.

Parameters:: data – Dictionary with content fields.
Returns:: FetchedContent instance.

metadata: dict[str, Any]

source: str = 'unknown'

title: str

to_dict() → dict[str, Any][source]

Convert to dictionary for ToolResult.

Returns:: Dictionary with all fields.

url: str

class yoker.tools.GuardType(*values)[source]

Bases: str, Enum

Functional type of a guardrailed string parameter.

PATH = 'path'

QUERY = 'query'

TEXT = 'text'

URL = 'url'

class yoker.tools.Guardrail[source]

Bases: ABC

Abstract base class for tool guardrails.

Guardrails validate tool parameters against permission boundaries before a tool is executed. Each tool type will have specific guardrail implementations (e.g., path restrictions for filesystem tools).

Example

class PathGuardrail(Guardrail):

def validate(self, tool_name: str, params: dict[str, Any]) -> ValidationResult:

path = params.get(“path”, “”) if not path.startswith(“/allowed”):

return ValidationResult(valid=False, reason=”Path not allowed”)

return ValidationResult(valid=True)

abstractmethod validate(tool_name: str, value: str | dict[str, Any]) → ValidationResult[source]

Validate tool parameters.

Parameters:

tool_name – Name of the tool being validated.
value – Either the extracted parameter value or the full params dict.

Returns:

ValidationResult indicating whether parameters are valid.

class yoker.tools.OllamaWebFetchBackend(async_client: AsyncClient, timeout_seconds: int = 30, max_size_kb: int = 2048)[source]

Bases: object

Web fetch backend using Ollama’s native web_fetch function.

Uses the Ollama Python SDK’s built-in web_fetch capability. Requires an authenticated AsyncClient for cloud-based fetch.

Features:

Native Ollama SDK integration
Built-in content extraction and summarization
Configurable output format

Limitations:

Requires backend.ollama.api_key for cloud-based fetch
Limited control over fetch process
Cannot enforce all client-side guardrails

async fetch(url: str, *, content_type: str = 'markdown', max_size_kb: int | None = None, timeout_seconds: int | None = None) → FetchedContent[source]

Fetch content via Ollama web_fetch function.

Uses client.web_fetch() which returns structured content.

Parameters:

url – URL to fetch.
content_type – Output format (default “markdown”).
max_size_kb – Max content size (uses default if None).
timeout_seconds – Timeout (uses default if None).

Returns:

FetchedContent with extracted content.

Raises:

WebFetchError – If Ollama request fails or content exceeds limits.

class yoker.tools.OllamaWebSearchBackend(async_client: AsyncClient, timeout_seconds: int = 30)[source]

Bases: object

Web search backend using Ollama’s native web_search function.

Uses the Ollama Python SDK’s built-in web_search capability. Requires an authenticated AsyncClient for cloud-based web search.

Features:

Native Ollama SDK integration
No model selection needed
Built-in result formatting

Limitations:

Requires backend.ollama.api_key for cloud-based search
Limited to 10 results
No domain filtering on client side

async search(query: str, max_results: int = 10) → list[SearchResult][source]

Execute search via Ollama web_search function.

Uses client.web_search() which returns structured results directly.

Parameters:

query – Search query string.
max_results – Maximum results (capped at 10 for Ollama).

Returns:

List of SearchResult objects.

Raises:

WebSearchError – If Ollama request fails.

class yoker.tools.Path(description: str = '', yoker_type: GuardType = GuardType.PATH)[source]

Bases: Text

Marker for filesystem path parameters.

yoker_type: GuardType = 'path'

class yoker.tools.PathGuardrail(config: Config)[source]

Bases: Guardrail

Concrete guardrail for filesystem tool validation.

Validates tool parameters against permission boundaries defined in Config: - Allowed filesystem paths (root containment) - Blocked regex patterns (e.g., .env, credentials) - Allowed file extensions (for read tool) - Maximum file size (for read tool)

Uses os.path.realpath() to resolve symlinks and normalize paths before validation, preventing path traversal attacks.

Example

guardrail = PathGuardrail(config) result = guardrail.validate(“read”, {“path”: “/etc/passwd”}) # result.valid is False because /etc/passwd is outside allowed paths

validate(tool_name: str, value: str | dict[str, Any]) → ValidationResult[source]

Validate tool parameters against permission boundaries.

Steps:

Skip non-filesystem tools immediately.
Extract and validate the path parameter.
Resolve the path to an absolute real path.
Check the path is within allowed roots.
Check blocked patterns.
For read tool: check extension and file size.

Parameters:

tool_name – Name of the tool being validated.
value – Either a path string or a dict of tool parameters. When called from _validate_tool_args, this is the extracted path string. When called directly in tests, this may be the full params dict.

Returns:

ValidationResult indicating whether parameters are valid.

class yoker.tools.Query(description: str = '', yoker_type: GuardType = GuardType.QUERY)[source]

Bases: Text

Marker for web search query parameters.

yoker_type: GuardType = 'query'

class yoker.tools.QueryWebGuardrail(config: WebGuardrailConfig | None = None)[source]: Bases: WebGuardrail

class yoker.tools.SearchResult(title: str, url: str, snippet: str, source: str = 'unknown')[source]

Bases: object

A single web search result.

title

Page title.

Type:: str

url

Result URL.

Type:: str

snippet

Short text snippet/summary.

Type:: str

source

Backend that produced this result (e.g., “ollama”, “duckduckgo”).

Type:: str

classmethod from_dict(data: dict[str, Any]) → SearchResult[source]

Create SearchResult from dictionary.

Parameters:: data – Dictionary with result fields.
Returns:: SearchResult instance.

snippet: str

source: str = 'unknown'

title: str

to_dict() → dict[str, str][source]

Convert SearchResult to dictionary.

Returns:: Dictionary with all fields.

url: str

class yoker.tools.Text(description: str = '', yoker_type: GuardType = GuardType.TEXT)[source]

Bases: object

Marker for plain text parameters with no guardrail.

description

Parameter description for the generated schema.

Type:: str

yoker_type

Functional type used by the guardrail dispatcher.

Type:: yoker.tools.annotations.GuardType

description: str = ''

yoker_type: GuardType = 'text'

class yoker.tools.ToolContext(config: ToolConfig, shared: ToolsSharedConfig, backends: dict[str, Any], session: Session | None = None)[source]

Bases: object

Execution context for tools.

Injected into tool functions that have a ctx: ToolContext parameter. Provides tool-specific config, shared settings, and backends.

config

Tool-specific config (WriteToolConfig, etc.)

Type:: ToolConfig

shared

Shared tool settings (content_display, etc.).

Type:: ToolsSharedConfig

backends

Provider-specific tool backends ({“websearch”: OllamaWebSearchBackend, …}).

Type:: dict[str, Any]

session

The Session owning the calling agent, when the agent runs inside a session. None on the single-agent path.

Type:: Session | None

backends: dict[str, Any]

config: ToolConfig

session: Session | None = None

shared: ToolsSharedConfig

class yoker.tools.ToolRegistry(dict=None, /, **kwargs)[source]

Bases: UserDict[str, ToolSpec]

Registry for managing available tools.

Tools are registered by passing a plain function or callable class instance. The registry stores the resulting ToolSpec and can be used like a normal dictionary.

Example

registry = ToolRegistry() registry.register(read_file) spec = registry[“read_file”] schemas = [s.schema for s in registry.values()]

find_tools(namespace: str) → list[ToolSpec][source]

get_schemas() → list[dict[str, Any]][source]

Return schemas for all registered tools.

Returns:: List of tool schemas in Ollama function format.

property names: list[str]: Return all registered tool names sorted alphabetically.

property namespaces: list[str]: Return all registered tool namespaces sorted alphabetically.

register(tool: Callable[[...], Any], *, namespace: str | None = None, name: str | None = None) → ToolSpec[source]

Register a callable as a tool.

Parameters:

tool – Function or callable class instance to register.
namespace – Optional namespace prefix for the tool name.
name – Optional explicit tool name override.

Returns:

The ToolSpec that was registered.

Raises:

ValueError – If a tool with the same name is already registered.

register_all(specs: list[ToolSpec], namespace: str) → None[source]

Register pre-built ToolSpec objects under a namespace.

Mirrors AgentRegistry.register_all(). Specs are already namespaced from plugin load.

register_plugin_tools(plugins: list[PluginComponents], config: Config) → None[source]

Register tools from clean plugin list, applying config-level filtering.

Consumes the generator output of load_plugins(). Security and global-enabled gating happen in load_plugins; only tool-level enabled/api-key filtering happens here.

property tools: list[ToolSpec]: Return all registered tool specs sorted by name.

class yoker.tools.ToolResult(success: bool, result: str | dict[str, Any] = '', error: str | None = None, content_metadata: dict[str, Any] | None = None)[source]

Bases: object

Result of a tool execution.

success

Whether the tool executed successfully.

Type:: bool

result

The result data (string content or dict for structured results).

Type:: str | dict[str, Any]

error

Error message if success is False.

Type:: str | None

content_metadata

Optional metadata for content display events. When provided, the agent emits a ToolContentEvent after ToolResultEvent. Contains operation, path, content_type, content, and metadata dict.

Type:: dict[str, Any] | None

content_metadata: dict[str, Any] | None = None

error: str | None = None

result: str | dict[str, Any] = ''

success: bool

class yoker.tools.Url(description: str = '', yoker_type: GuardType = GuardType.URL)[source]

Bases: Text

Marker for URL parameters.

yoker_type: GuardType = 'url'

class yoker.tools.UrlWebGuardrail(config: WebGuardrailConfig | None = None)[source]

Bases: WebGuardrail

validate(tool_name: str, value: str | dict[str, str]) → ValidationResult[source]

Validate a URL for web fetch.

Steps:

Validate URL format (scheme, host, etc.).
Check for SSRF attempts (private IPs, metadata endpoints).
Check domain allowlist if configured.
Check domain blocklist if configured.
Check HTTPS requirement if configured.

Parameters:: url – URL string to validate.
Returns:: ValidationResult with success/failure and reason.

class yoker.tools.ValidationResult(valid: bool, reason: str | None = None)[source]

Bases: object

Result of a guardrail validation check.

valid

Whether the parameters passed validation.

Type:: bool

reason

Explanation if validation failed.

Type:: str | None

reason: str | None = None

valid: bool

class yoker.tools.WebFetchBackend(*args, **kwargs)[source]

Bases: Protocol

Protocol for web fetch backend implementations.

Defines the interface that all fetch backends must implement. Supports async native tools.

Implementations:

OllamaWebFetchBackend: Uses Ollama’s native web_fetch function

async fetch(url: str, *, content_type: str = 'markdown', max_size_kb: int = 2048, timeout_seconds: int = 30) → FetchedContent[source]

Fetch content from a URL.

Parameters:

url – URL to fetch.
content_type – Output format (“markdown”, “text”, “html”).
max_size_kb – Maximum content size in KB.
timeout_seconds – Fetch timeout in seconds.

Returns:

FetchedContent with extracted content.

Raises:

WebFetchError – If fetch fails.

exception yoker.tools.WebFetchError(message: str, url: str = '', backend: str = 'unknown', cause: Exception | None = None, error_type: str = 'unknown')[source]

Bases: Exception

Exception for web fetch errors.

message: Human-readable error message.

url: URL that failed (if applicable).

backend: Backend that raised the error.

cause: Original exception if wrapped.

error_type: Type of error (ssrf, timeout, size, invalid_url, etc.).

class yoker.tools.WebGuardrail(config: WebGuardrailConfig | None = None)[source]

Bases: Guardrail

Guardrail for web tool validation.

Validates:

Query length (prevents excessive queries)
Domain allowlist (optional, for restricted searches)
Domain blocklist (optional, for blocked domains)
SSRF protection (blocks private IPs, cloud metadata)
Query sanitization (blocks sensitive patterns)
Rate limiting (requests per minute, per hour, concurrent)

Note

Domain filtering is client-side validation only. Ollama backend may still access blocked domains. For full control, use LocalWebSearchBackend.

release_concurrent(user_id: str = 'default') → None[source]

Release a concurrent request slot.

Call this after search completes to decrement concurrent count.

Parameters:: user_id – User/session identifier.

validate(tool_name: str, value: str | dict[str, str]) → ValidationResult[source]

Validate web search parameters.

Steps:

Validate query is present and non-empty.
Validate query length <= max_query_length.
Check for SSRF attempts (private IPs, cloud metadata).
Check domain allowlist if configured.
Check domain blocklist if configured.
Check for sensitive patterns in query.
Check rate limits.

Parameters:

tool_name – Name of tool being validated.
value – Either a query string or a dict with ‘query’ key.

Returns:

ValidationResult with success/failure and reason.

class yoker.tools.WebGuardrailConfig(max_query_length: int = 500, domain_allowlist: tuple[str, ...] = (), domain_blocklist: tuple[str, ...] = (), requests_per_minute: int = 60, requests_per_hour: int = 1000, max_concurrent_requests: int = 0, block_private_cidrs: bool = True, timeout_seconds: int = 30, require_https: bool = True)[source]

Bases: object

Configuration for WebGuardrail.

max_query_length

Maximum query string length (default 500).

Type:: int

domain_allowlist

Domains to allow (empty = all allowed).

Type:: tuple[str, …]

domain_blocklist

Domains to block (empty = none blocked).

Type:: tuple[str, …]

requests_per_minute

Maximum requests per minute (0 = unlimited).

Type:: int

requests_per_hour

Maximum requests per hour (0 = unlimited).

Type:: int

max_concurrent_requests

Maximum concurrent requests (0 = unlimited).

Type:: int

block_private_cidrs

Whether to block private IP ranges.

Type:: bool

timeout_seconds

Search timeout in seconds.

Type:: int

require_https

Whether to require HTTPS URLs (block HTTP).

Type:: bool

block_private_cidrs: bool = True

domain_allowlist: tuple[str, ...] = ()

domain_blocklist: tuple[str, ...] = ()

max_concurrent_requests: int = 0

max_query_length: int = 500

requests_per_hour: int = 1000

requests_per_minute: int = 60

require_https: bool = True

timeout_seconds: int = 30

class yoker.tools.WebSearchBackend(*args, **kwargs)[source]

Bases: Protocol

Protocol for web search backend implementations.

Defines the interface that all search backends must implement. Supports async native tools.

Implementations:

OllamaWebSearchBackend: Uses Ollama’s native web_search function

async search(query: str, max_results: int = 10) → list[SearchResult][source]

Execute a web search and return results.

Parameters:

query – Search query string.
max_results – Maximum number of results to return (1-50).

Returns:

List of SearchResult objects.

Raises:

WebSearchError – If search fails.

exception yoker.tools.WebSearchError(message: str, backend: str = 'unknown', cause: Exception | None = None)[source]

Bases: Exception

Base exception for web search errors.

message: Human-readable error message.

backend: Backend that raised the error.

cause: Original exception if wrapped.

API Reference

Current Modules

yoker.api

yoker.core

yoker.session

yoker.events

yoker.ui

yoker.tools

`yoker.api`

`yoker.core`

`yoker.session`

`yoker.events`

`yoker.ui`

`yoker.tools`