API Reference

API documentation will be added as implementation progresses.

Current Modules

`yoker.agent`

Asynchronous Agent implementation for Yoker.

This module provides Agent, an async-first agent that chats with Ollama and uses tools. All I/O operations are async.

Bases: object

Asynchronous agent that chats with Ollama and uses tools.

This implementation uses composition with AgentCore for shared state. All I/O operations are async.

client: AsyncClient for Ollama API communication.

model: Model to use for chat.

config: Configuration object.

context: ContextManager for conversation history.

thinking_mode: Current thinking mode (on/off/silent).

agent_definition: Loaded agent definition (if provided).

_recursion_depth: Current recursion depth (internal, for subagent tracking).

_max_recursion_depth: Maximum allowed recursion depth (internal).

add_event_handler(handler: Callable[[Event], None] | Callable[[Event], Coroutine[None, None, None]]) → None[source]

Register an event handler.

Event handlers receive all events emitted during agent processing. Handlers can access potentially sensitive data (tool results, file contents). Only register handlers from trusted sources.

Supports both sync and async handlers:

Sync handlers: def handler(event: Event) -> None
Async handlers: async def handler(event: Event) -> None

Parameters:: handler – Callable that receives Event objects.

Example

def my_handler(event: Event):

if isinstance(event, ContentChunkEvent):: print(event.text, end=’’, flush=True)

agent.add_event_handler(my_handler)

Security Note (SEC-ASYNC-1):: Handler registration is logged for audit purposes.

property agent_definition: AgentDefinition | None: Loaded agent definition, if any.

async begin_session() → None[source]

Begin an agent session asynchronously.

Emits SESSION_START event with session metadata. Saves context to persist session state. Call this before processing messages.

property client: AsyncClient: AsyncClient for Ollama API communication.

property command_registry: CommandRegistry | None: Command registry for slash-commands.

property config: Config: Configuration object.

property context: ContextManager: Context manager for conversation history.

async end_session(reason: str = 'quit') → None[source]

End an agent session asynchronously.

Emits SESSION_END event. Closes context to ensure all data is persisted. Call this when done processing messages.

Parameters:: reason – Reason for ending the session (e.g., “quit”, “error”, “interrupt”).

property model: str: Model name to use for chat.

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

Process a single message and return the response asynchronously.

Handles tool calls internally until a final response is ready. Uses streaming when thinking is enabled.

Emits events during processing: - TURN_START - THINKING_START/CHUNK/END (if enabled) - CONTENT_START/CHUNK/END - TOOL_CALL/RESULT (if tools called) - TURN_END

Parameters:: message – User message to process.
Returns:: Assistant’s response text.
Raises:: NetworkError – If communication with Ollama fails.

remove_event_handler(handler: Callable[[Event], None] | Callable[[Event], Coroutine[None, None, None]]) → None[source]

Remove a registered event handler.

Parameters:: handler – The handler to remove.
Raises:: ValueError – If the handler is not registered.

property thinking_mode: ThinkingMode: Current thinking mode state.

property tool_registry: ToolRegistry: Registry of available tools.

`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.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.ConsoleEventHandler(console: Console | None = None, show_thinking: bool = True, show_tool_calls: bool = True, wrap_width: int | None = None, version: str = '0.1.0')[source]

Bases: object

Handles events by rendering to Rich console.

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

Bases: Event

Emitted for each chunk of content output.

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.ErrorEvent(error_type: str, message: str, details: dict[str, ~typing.Any]=<factory>, *, type: EventType, timestamp: datetime = <factory>)[source]

Bases: Event

Emitted when an error occurs.

details: dict[str, Any]

error_type: str

message: str

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.EventHandler(*args, **kwargs)[source]

Bases: Protocol

Protocol for event handlers.

Event handlers receive all events emitted during agent processing. Both sync and async handlers are supported:

Sync handlers: def __call__(self, event: Event) -> None
Async handlers: async def __call__(self, event: Event) -> None

Handlers should complete quickly (ideally <100ms) to avoid blocking the event loop. For I/O-bound operations, use async handlers.

Security Note:: Handlers can access potentially sensitive data (tool results, file contents). Only register handlers from trusted sources.

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.add_event_handler(recorder) # … run session … recorder.close()

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

class yoker.events.EventReplayAgent(events_path: Path)[source]

Bases: object

Agent that replays events from a JSONL file.

This class provides the same async interface as Agent but replays previously recorded events instead of calling the LLM. Useful for:

Generating screenshots without LLM costs
Testing event handlers
Debugging session flows

Example

agent = EventReplayAgent(Path(“session.jsonl”)) agent.add_event_handler(ConsoleEventHandler(console)) await agent.process(“Hello”) # Replays events for “Hello” turn

add_event_handler(handler: EventCallback) → None[source]

Register an event handler for replay.

Parameters:: handler – Callable that receives Event objects.

async begin_session() → None[source]: No-op for replay agent - session already in event log.

async end_session(reason: str = 'quit') → None[source]: No-op for replay agent.

property model: str: Return the model name from the recorded session.

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

Replay events for one turn.

Finds the matching TURN_START event and replays all events until TURN_END, emitting them to registered handlers.

Parameters:: message – The user message (used to find matching turn).
Returns:: The response text from the replayed turn.

async replay_command(command: str) → str[source]

Replay a command event.

Finds the matching COMMAND event and emits it to handlers.

Parameters:: command – The command string (used to find matching event).
Returns:: The command result, or empty string if not found.

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

Bases: Enum

Enumeration of all event types.

COMMAND = 14

CONTENT_CHUNK = 9

CONTENT_END = 10

CONTENT_START = 8

ERROR = 15

SESSION_END = 2

SESSION_START = 1

THINKING_CHUNK = 6

THINKING_END = 7

THINKING_START = 5

TOOL_CALL = 11

TOOL_CONTENT = 13

TOOL_RESULT = 12

TURN_END = 4

TURN_START = 3

class yoker.events.LiveDisplay(console: Console | None = None, refresh_per_second: int = 4)[source]

Bases: object

Live-updating display for streaming content with spinner.

Uses Rich’s Live to manage a continuously refreshing display that shows: - Thinking content (dimmed style) - Response content - Status spinner (during streaming) - Turn statistics (after completion)

Example

with LiveDisplay() as display:: display.append_thinking(“Thinking…”) display.append_response(“Hello”) display.show_stats(tokens=100, duration_ms=1500)

append_response(text: str) → None[source]

Append text to the response content area.

Parameters:: text – Text to append (normal style).

append_thinking(text: str) → None[source]

Append text to the thinking content area.

Parameters:: text – Text to append (displayed dimmed).

clear() → None[source]: Clear all content and reset for next turn.

show_stats(prompt_tokens: int = 0, eval_tokens: int = 0, duration_ms: int = 0) → None[source]

Show turn statistics instead of spinner.

Parameters:

prompt_tokens – Number of prompt/input tokens.
eval_tokens – Number of generated/output tokens.
duration_ms – Total duration in milliseconds.

start_spinner() → None[source]

Show the spinner.

Creates the spinner if it doesn’t exist and activates it. Called automatically when content starts streaming.

stop_spinner() → None[source]

Remove the spinner from the display.

Called before exiting the Live context to prevent the spinner from showing in the final output.

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

Bases: Event

Emitted when agent session ends.

reason: str

class yoker.events.SessionStartEvent(model: str, thinking_enabled: bool, config_summary: dict[str, ~typing.Any]=<factory>, *, type: EventType, timestamp: datetime = <factory>)[source]

Bases: Event

Emitted when agent session starts.

config_summary: dict[str, Any]

model: str

thinking_enabled: bool

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).

This event is optional - tools can emit it when they have meaningful content to show. The event handler can choose to display or ignore it.

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

Type of content (“full”, “diff”, “summary”).

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, 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

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[source]

Deserialize a dictionary back to an event object.

Parameters:

entry – Dictionary with type, timestamp, and event data.

Returns:

Reconstructed event object.

Raises:

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

yoker.events.live_display(console: Console | None = None, refresh_per_second: int = 4) → Iterator[LiveDisplay][source]

Context manager for live display.

Parameters:

console – Rich console (default: new Console).
refresh_per_second – Refresh rate for Live display.

Yields:

LiveDisplay instance for content updates.

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

Serialize an event to a JSON-serializable dictionary.

Parameters:: event – The event to serialize.
Returns:: Dictionary with type, timestamp, and event data.

Event Types:

Event	Description
`SessionStartEvent`	Emitted when agent session begins
`SessionEndEvent`	Emitted when agent session ends
`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
`ErrorEvent`	Emitted when an error occurs

EventHandler Protocol:

from yoker.events import EventHandler, Event

class MyHandler:
    def __call__(self, event: Event) -> None:
        if hasattr(event, 'text'):
            print(event.text)

agent.add_event_handler(MyHandler())

`yoker.tools`

Tools package for Yoker.

Provides the tool framework including base classes, registry, guardrails, and concrete tool implementations.

class yoker.tools.ExistenceTool(guardrail: Guardrail | None = None)[source]

Bases: Tool

Tool for checking file or folder existence.

Checks whether a file or directory exists at the given path. When a guardrail is provided, validates parameters before checking. Resolves paths with realpath and rejects symlinks by default.

Returns a structured result indicating existence, type (file/directory), and the resolved path for debugging.

property description: str: Tool description shown to the LLM.

async execute(**kwargs: Any) → ToolResult[source]

Check if a file or folder exists.

Steps:

Validate parameters via guardrail if provided.
Validate path parameter is a non-empty string.
Reject symlinks before resolving.
Resolve the path with os.path.realpath().
Check existence and type (file or directory).
Return structured result with boolean existence flag.

Parameters:: **kwargs – Must contain ‘path’ key with path to check.
Returns:: ToolResult with existence check result or error message.

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

Return Ollama-compatible schema for the existence tool.

Returns:: ‘function’ and function metadata.
Return type:: Dict with ‘type’

property name: str: Tool name used for registration and LLM tool calling.

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.GitTool(config: GitToolConfig, guardrail: Guardrail | None = None, permission_handlers: dict[str, HandlerConfig] | None = None)[source]

Bases: Tool

Tool for executing Git operations with security guardrails.

Provides controlled access to Git commands through an operation allowlist. Destructive operations (commit, push) require explicit permission handling. All commands are executed via subprocess with list arguments to prevent shell injection.

_config: GitToolConfig with allowed commands and permission requirements.

_permission_handlers: Dict of operation name to HandlerConfig.

property description: str: Tool description shown to the LLM.

async execute(**kwargs: Any) → ToolResult[source]

Execute a Git operation.

Steps:

Extract and validate operation parameter.
Validate repository path via guardrail.
Check permission for destructive operations.
Build and sanitize command.
Execute command via subprocess.
Format and return result.

Parameters:: **kwargs – Must contain ‘operation’ key, optional ‘path’ and ‘args’.
Returns:: ToolResult with command output or error message.

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

Return Ollama-compatible schema for the git tool.

Returns:: ‘function’ and function metadata.
Return type:: Dict with ‘type’

property name: str: Tool name used for registration and LLM tool calling.

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, params: dict[str, Any]) → ValidationResult[source]

Validate tool parameters.

Parameters:

tool_name – Name of the tool being validated.
params – Dictionary of tool parameters from the LLM.

Returns:

ValidationResult indicating whether parameters are valid.

class yoker.tools.ListTool(guardrail: Guardrail | None = None)[source]

Bases: Tool

Tool for listing directory contents.

Lists files and directories with optional recursion depth control, entry limits, and glob pattern filtering. Returns a tree-formatted string for LLM consumption.

When a guardrail is provided, validates parameters before listing.

ABSOLUTE_MAX_DEPTH: int = 10

ABSOLUTE_MAX_ENTRIES: int = 5000

DEFAULT_MAX_DEPTH: int = 1

DEFAULT_MAX_ENTRIES: int = 1000

property description: str: Tool description shown to the LLM.

async execute(**kwargs: Any) → ToolResult[source]

List directory contents with optional filtering and limits.

Parameters:: **kwargs – Must contain ‘path’. May contain ‘max_depth’, ‘max_entries’, and ‘pattern’.
Returns:: ToolResult with formatted directory listing or error message.

get_schema() → dict[str, Any][source]: Return Ollama-compatible schema for the list tool.

property name: str: Tool name used for registration and LLM tool calling.

class yoker.tools.MkdirTool(guardrail: Guardrail | None = None)[source]

Bases: Tool

Tool for creating directories.

Creates a directory at the given path with optional recursive parent creation. When a guardrail is provided, validates parameters before creating. Resolves paths with realpath and rejects symlinks by default.

Returns a structured result indicating success, creation status, and the resolved path for debugging.

property description: str: Tool description shown to the LLM.

async execute(**kwargs: Any) → ToolResult[source]

Create a directory.

Steps:

Validate parameters via guardrail if provided.
Extract and validate path parameter.
Reject symlinks before resolving.
Resolve the path with os.path.realpath().
Check if path already exists (file or directory).
Create directory, optionally with parents.
Return structured result with creation status.

Parameters:: **kwargs – Must contain ‘path’ key. May contain ‘recursive’ (default False).
Returns:: ToolResult with creation result or error message.

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

Return Ollama-compatible schema for the mkdir tool.

Returns:: ‘function’ and function metadata.
Return type:: Dict with ‘type’

property name: str: Tool name used for registration and LLM tool calling.

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 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 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.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, params: 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.
params – Dictionary of tool parameters from the LLM.

Returns:

ValidationResult indicating whether parameters are valid.

class yoker.tools.ReadTool(guardrail: Guardrail | None = None)[source]

Bases: Tool

Tool for reading file contents.

Reads the entire contents of a file as text with defense-in-depth validation. When a guardrail is provided, validates parameters before reading. Resolves paths with realpath, rejects symlinks by default, and reads with explicit UTF-8 encoding.

Error messages returned to the LLM are sanitized to avoid leaking filesystem structure. Full paths are logged internally for debugging.

property description: str: Tool description shown to the LLM.

async execute(**kwargs: Any) → ToolResult[source]

Read a file and return its contents.

Steps:

Validate parameters via guardrail if provided.
Resolve the path with os.path.realpath().
Reject symlinks unless explicitly allowed.
Verify the file exists.
Read with UTF-8 encoding and replacement for invalid bytes.
Log access for audit trail.

Parameters:: **kwargs – Must contain ‘path’ key with file path.
Returns:: ToolResult with file content or sanitized error message.

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

Return Ollama-compatible schema for the read tool.

Returns:: ‘function’ and function metadata.
Return type:: Dict with ‘type’

property name: str: Tool name used for registration and LLM tool calling.

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.SearchTool(guardrail: Guardrail | None = None)[source]

Bases: Tool

Tool for searching files and their contents.

Supports two search modes: - ‘content’: Search within file contents (grep-like, using regex) - ‘filename’: Search file names (glob-like, using fnmatch)

All searches respect allowed paths guardrails and enforce limits to prevent resource exhaustion.

ReDoS Prevention: - Pattern length limit (500 characters) - Forbidden pattern detection (nested quantifiers) - Compile-time regex validation - File size filtering (skip large files)

ABSOLUTE_MAX_RESULTS: int = 1000

ABSOLUTE_TIMEOUT_MS: int = 30000

DEFAULT_MAX_RESULTS: int = 100

DEFAULT_TIMEOUT_MS: int = 5000

FORBIDDEN_PATTERNS: tuple[str, ...] = ('\\([^)]*[+*][^)]*\\)[+*]', '\\([^)]*\\|[^)]*\\)[+*]')

MAX_FILE_SIZE_KB: int = 500

MAX_PATTERN_LENGTH: int = 500

SKIP_DIRS: frozenset[str] = frozenset({'*.egg-info', '.eggs', '.git', '.mypy_cache', '.pytest_cache', '.tox', '.venv', '__pycache__', 'build', 'dist', 'htmlcov', 'node_modules', 'venv'})

property description: str: Tool description shown to the LLM.

async execute(**kwargs: Any) → ToolResult[source]

Search files with pattern matching and limits.

Parameters:: **kwargs – Must contain ‘path’. May contain ‘pattern’, ‘type’, and ‘max_results’.
Returns:: ToolResult with search results or error message.

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

Return Ollama-compatible schema for the search tool.

Returns:: ‘function’ and function metadata.
Return type:: Dict with ‘type’

property name: str: Tool name used for registration and LLM tool calling.

class yoker.tools.Tool(guardrail: Guardrail | None = None)[source]

Bases: ABC

Abstract base class for all Yoker tools.

Each tool must define its name, description, JSON schema for the LLM, and an execute method that returns a ToolResult.

Tools may optionally accept a Guardrail instance for defense-in-depth validation. When a guardrail is provided, the tool validates parameters in execute() before performing any I/O.

Example

class MyTool(Tool):

@property def name(self) -> str:

return “my_tool”

@property def description(self) -> str:

return “Does something useful”

def get_schema(self) -> dict[str, Any]:

return {

“type”: “function”, “function”: {

“name”: self.name, “description”: self.description, “parameters”: {

“type”: “object”, “properties”: {

“arg”: {“type”: “string”}

}, “required”: [“arg”]

}

}

async def execute(self, arg: str) -> ToolResult:

return ToolResult(success=True, result=f”Got: {arg}”)

abstract property description: str: Tool description shown to the LLM.

abstractmethod async execute(**kwargs: Any) → ToolResult[source]

Execute the tool with the given parameters.

All tools must implement async execution for proper async/await integration with the agent’s event loop.

Parameters:: **kwargs – Parameters from the LLM tool call.
Returns:: ToolResult with success status and output.

exists(path: str) → bool[source]

Check if a file exists, validating through the guardrail if available.

This is a utility method for tools that need to check file existence as part of their operations (e.g., ReadTool before reading).

Parameters:: path – The file path to check.
Returns:: True if the path exists and passes guardrail validation, False otherwise (including if guardrail rejects the path).

abstractmethod get_schema() → dict[str, Any][source]

Return Ollama-compatible function schema.

Returns:

{

“type”: “function”, “function”: {

”name”: “…”, “description”: “…”, “parameters”: {…}

}

Return type:

Dict in OpenAI function-calling format

abstract property name: str: Tool name used for registration and LLM tool calling.

class yoker.tools.ToolRegistry[source]

Bases: object

Registry for managing available tools.

Tools are registered by name and can be retrieved for execution or schema generation for the LLM.

_tools: Internal dictionary mapping tool names to Tool instances.

Example

registry = ToolRegistry() registry.register(ReadTool()) tool = registry.get(“read”) schemas = registry.get_schemas()

exists(tool_name: str, path: str) → bool[source]

Check if a file exists using a tool’s guardrail validation.

Delegates to the tool’s exists method which validates through the guardrail before checking file existence.

Parameters:

tool_name – Name of the tool to use for validation.
path – File path to check.

Returns:

True if the tool exists and the path exists after validation, False if tool not found, guardrail rejects, or file doesn’t exist.

get(name: str) → Tool | None[source]

Get a tool by name.

Parameters:: name – Tool name (case-sensitive).
Returns:: The Tool instance if found, None otherwise.

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

Get schemas for all registered tools.

Returns:: List of Ollama-compatible function schemas.

list_tools() → list[Tool][source]

Get all registered tools.

Returns:: List of Tool instances sorted by name.

property names: list[str]

Get all registered tool names.

Returns:: List of tool names sorted alphabetically.

register(tool: Tool) → None[source]

Register a tool.

Parameters:: tool – The Tool instance to register.
Raises:: ValueError – If a tool with the same name is already registered.

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.UpdateTool(guardrail: Guardrail | None = None, config: Config | None = None)[source]

Bases: Tool

Tool for updating existing file contents.

Edits files via replace, insert_before, insert_after, or delete operations. When a guardrail is provided, validates parameters before updating. Resolves paths with realpath, rejects symlinks, and enforces exact match validation and diff size limits via config.

Error messages returned to the LLM are sanitized to avoid leaking filesystem structure. Full paths are logged internally for debugging.

property description: str: Tool description shown to the LLM.

async execute(**kwargs: Any) → ToolResult[source]

Update an existing file.

Steps:

Validate parameters via guardrail if provided.

Extract and validate path, operation, and content parameters.

Resolve the path with os.path.realpath().

Reject symlinks unless explicitly allowed.

Verify file exists and is a file (not a directory).

Read file fresh to prevent TOCTOU race conditions.

Validate diff size against config limit.

Perform the requested operation with exact match validation.

Write atomically via temp file + os.replace().

Log update for audit trail.

Parameters:: **kwargs – Must contain ‘path’ and ‘operation’ keys. May contain ‘old_string’, ‘new_string’, ‘line_number’.
Returns:: ToolResult with success status and output or error message.

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

Return Ollama-compatible schema for the update tool.

Returns:: ‘function’ and function metadata.
Return type:: Dict with ‘type’

property name: str: Tool name used for registration and LLM tool calling.

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.WebFetchTool(backend: WebFetchBackend | None = None, guardrail: WebGuardrail | None = None)[source]

Bases: Tool

Tool for fetching web content using pluggable backends.

Fetches content from URLs and returns structured results. Uses a configurable backend (Ollama native or local httpx). Validates URLs through WebGuardrail before execution.

Example

tool = WebFetchTool(backend=OllamaWebFetchBackend(async_client)) result = await tool.execute(url=”https://example.com”, content_type=”markdown”)

property description: str: Tool description shown to the LLM.

async execute(**kwargs: Any) → ToolResult[source]

Execute web fetch with the given parameters asynchronously.

Steps:

Validate parameters via guardrail if provided.
Extract and validate URL parameter.
Delegate to backend for fetch execution.
Return structured results or error.

Parameters:: **kwargs – Must contain ‘url’, optionally ‘content_type’, ‘max_size_kb’.
Returns:: ToolResult with FetchedContent dict or error.

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

Return Ollama-compatible schema.

Returns:: Schema with url, content_type, and max_size_kb parameters.

property name: str: Tool name used for registration.

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, params: dict[str, Any]) → 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.
params – Tool parameters from LLM.

Returns:

ValidationResult with success/failure and reason.

validate_url(url: 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.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.

class yoker.tools.WebSearchTool(backend: WebSearchBackend | None = None, guardrail: WebGuardrail | None = None)[source]

Bases: Tool

Tool for searching the web using pluggable backends.

Searches the web for information and returns structured results. Uses a configurable backend (Ollama native or local DDGS). Validates queries through WebGuardrail before execution.

Example

tool = WebSearchTool(backend=OllamaWebSearchBackend()) result = await tool.execute(query=”Python async best practices”, max_results=5)

property description: str: Tool description shown to the LLM.

async execute(**kwargs: Any) → ToolResult[source]

Execute web search with the given parameters asynchronously.

Steps:

Validate parameters via guardrail if provided.
Extract and validate query parameter.
Delegate to backend for search execution.
Return structured results or error.

Parameters:: **kwargs – Must contain ‘query’, optionally ‘max_results’.
Returns:: ToolResult with list of SearchResult dicts or error.

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

Return Ollama-compatible schema.

Returns:: Schema with query and max_results parameters.

property name: str: Tool name used for registration.

class yoker.tools.WriteTool(guardrail: Guardrail | None = None, config: Config | None = None)[source]

Bases: Tool

Tool for writing file contents.

Writes content to a file with defense-in-depth validation. When a guardrail is provided, validates parameters before writing. Resolves paths with realpath, rejects symlinks, and supports overwrite protection and parent directory creation.

Error messages returned to the LLM are sanitized to avoid leaking filesystem structure. Full paths are logged internally for debugging.

property description: str: Tool description shown to the LLM.

async execute(**kwargs: Any) → ToolResult[source]

Write content to a file.

Steps:

Validate parameters via guardrail if provided.
Extract and validate path and content parameters.
Resolve the path with os.path.realpath().
Reject symlinks unless explicitly allowed.
Check overwrite protection (config-based).
Create parent directories if requested.
Write with UTF-8 encoding.
Log write for audit trail.
Populate content_metadata for content display.

Parameters:: **kwargs – Must contain ‘path’ and ‘content’ keys. May contain ‘create_parents’ (default False).
Returns:: ToolResult with success status and output or error message.

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

Return Ollama-compatible schema for the write tool.

Returns:: ‘function’ and function metadata.
Return type:: Dict with ‘type’

property name: str: Tool name used for registration and LLM tool calling.

yoker.tools.create_default_registry(parent_agent: Agent | None = None) → ToolRegistry[source]

Create a registry with all built-in tools registered.

Parameters:: parent_agent – Optional parent agent for AgentTool (required for subagent spawning).
Returns:: ToolRegistry with default tools (read, list, write, update, search, existence, mkdir, git, agent).

API Reference

Current Modules

yoker.agent

yoker.events

yoker.tools

`yoker.agent`

`yoker.events`

`yoker.tools`