API Reference: Agents

agents/loop.py - AgentLoop

Purpose: Orchestrates the ReAct agent loop

AgentLoop(..., memory_retriever: MemoryRetrieverProtocol, skills_retriever: SkillsRetrieverProtocol | None = None, ...)

class AgentLoop:
    def __init__(
        self,
        llm_client: LocalLLMClient,
        context_builder: ContextBuilder,
        session_store: SessionStore,
        memory_retriever: MemoryRetrieverProtocol,
        tool_executor: ToolExecutor,
        app_config: AppConfig,
        working_memory_compressor: WorkingMemoryCompressor | None = None,
        skills_retriever: SkillsRetrieverProtocol | None = None,
        skills_on_event: AgentEventCallback | None = None,
        ...
    )
    
    async def run(
        self,
        user_id: str,
        session_id: str,
        user_input: str,
        agent_config: AgentConfig,
        *,
        resume: bool = False,
        memory_scope: str | None = None,
        trace_id: str | None = None,
        cancel_event: asyncio.Event | None = None,
        on_event: Callable[..., Awaitable[None]] | None = None,
    ) -> AgentRunResult:
        """Run the agent ReAct loop.

        Returns:
            AgentRunResult with `response`, `trace_id`, and `warnings`
        """

Agent Loop Flow

1. SessionStore.append(user message)
2. memory_retriever.retrieve(query, user_id, categories) — no-op when `NullMemoryRetriever`
3. skills_retriever.retrieve(query, user_id, max_tokens) — when `[skills]` enabled
4. Load L1 working memory (when working_memory_enabled and compressor wired)
5. history = SessionStore.load(session_id, user_id, limit=history_limit)
6. For iteration in range(max_steps):
   a. context = context_builder.build(
        system_prompt, tools, memory=RAG, skills=skills_context, history, user_input,
        working_memory=L1,
    )
   b. response = llm_client.chat(context)
   c. If tool call: ToolExecutor.execute → append observation to history
   d. Else: finalize
7. SessionStore.append(assistant response)
8. WorkingMemoryCompressor.update_summary(run_messages)  # when enabled
9. SessionStore.compact(keep_last) if message count > history_limit

agents/tool_parser.py - ToolParser

Purpose: Parses tool calls from LLM responses

ToolParser()

class ToolParser:
    def __init__(self)
    
    @staticmethod
    def parse_tool_call(text: str) -> tuple[str, dict] | None
        """Parse tool call from text.
        
        Accepts:
        - Raw JSON: {"tool": "name", "args": {...}}
        - Fenced code: ```json\n{"tool": "name", "args": {...}}\n```
        
        Returns:
            (tool_name, args_dict) or None if not a tool call
        """

Parsing Rules

Valid formats:

{"tool": "mcp_filesystem_read_file", "args": {"path": "test.txt"}}

{
  "tool": "mcp_filesystem_read_file",
  "args": {
    "path": "test.txt"
  }
}

Invalid formats:

mcp_filesystem_read_file with path="test.txt"
{"tool": "mcp_filesystem_read_file" path: "test.txt"}  # Invalid JSON

Fallback:
If parsing fails with non-JSON text, treat as final answer (no tool call). Salvageable JSON failures re-prompt the model.

agents/registry.py - Tool Executor

Purpose: Manages tool registration and execution

ToolExecutor(tools: list[Tool], timeout: float = 30.0, ...)

class ToolExecutor:
    def __init__(
        self,
        tools: list[Tool],
        timeout: float = 30.0,
        max_concurrent: int = 5,
        *,
        memory_manager: MemoryManager | None = None,
        tool_result_inline_limit: int = 6144,
    )
    
    async def execute(
        self,
        name: str,
        args: dict,
        *,
        allowed: frozenset[str] | None = None,
    ) -> ToolResult
        """Execute a tool. When `allowed` is set, rejects tools not in the set."""
    
    def list_tools(self, allowed: frozenset[str] | None = None) -> str
        """List available tools; filtered by `allowed` when provided."""

Tool Dataclass

@dataclass
class Tool:
    name: str                    # Unique tool name
    description: str             # Human-readable description
    input_schema: dict           # JSON Schema for validation
    func: Callable              # Async function to execute

Native and MCP tools

Capability tools come from host mcp.json as mcp_{server}_{tool} (auto-granted). Core registers only meta/recovery tools:

See docs/tools.md and docs/mcp.md. Built-in filesystem/shell/git/HTTP packages were removed — do not reintroduce them.

core/protocols.py — Memory and skills retrievers

Default wiring uses null implementations; optional packages replace them when installed and enabled.

class MemoryRetrieverProtocol(Protocol):
    async def retrieve(
        self, query: str, user_id: str, categories: list[str], max_tokens: int
    ) -> str: ...

class NullMemoryRetriever:
    async def retrieve(...) -> str:
        return ""

class SkillsRetrieverProtocol(Protocol):
    async def retrieve(self, query: str, user_id: str, max_tokens: int) -> str: ...

class NullSkillsRetriever:
    async def retrieve(...) -> str:
        return ""

Real MemoryRetriever lives in packages/local-ai-memory/memory/retriever.py when [memory] is enabled. Real SkillsRetriever lives in packages/local-ai-skills/local_ai_skills/store.py when [skills] is enabled.

memory/retriever.py - MemoryRetriever (optional [memory])

Purpose: 3-stage memory retrieval (BM25/hybrid → LLM select → load)

MemoryRetriever(memory_store: MemoryStore, llm_client: LocalLLMClient, token_counter: TokenCounter)

class MemoryRetriever:
    def __init__(
        self,
        memory_store: MemoryStore,
        llm_client: LocalLLMClient,
        token_counter: TokenCounter,
    )
    
    async def retrieve(
        self,
        query: str,
        user_id: str,
        categories: list[str],
        max_tokens: int = 2000,
    ) -> str
        """Retrieve relevant memory content.
        
        Uses 3-stage retrieval:
        1. BM25 or hybrid index search (no LLM)
        2. LLM file selection
        3. Content loading within token budget
        
        Args:
            query: User query
            user_id: User identifier
            categories: Memory categories to search
            max_tokens: Maximum tokens to load
            
        Returns:
            Concatenated memory content
        """

3-Stage Retrieval

Stage 1: BM25 / Hybrid

async def bm25_search(query: str, top_n: int = 5) -> list[str]:
    """Rank index entries via BM25Ranker. fast_filter() is an alias."""

async def hybrid_search(query: str, top_n: int = 5, bm25_weight: float = 0.5) -> list[str]:
    """BM25 + embedding vectors fused via RRF when memory.search_mode is hybrid."""

Stage 2: LLM File Selection

async def select_files(query: str, candidates: list[str]) -> list[str]:
    """LLM selects most relevant files from candidates.
    
    Sends prompt with candidate metadata to LLM.
    LLM returns JSON array of filenames to load.
    
    Args:
        query: User query
        candidates: List of candidate filenames
        
    Returns:
        List of selected filenames (max 5)
    """

MemoryStore Interface

class MemoryStore:
    def __init__(self, guard: DirectoryGuard)
    
    def list_user_files(self, user_id: str) -> list[str]
        """List all user files in category."""
    
    def read_file(self, path: str) -> str
        """Read file content."""
    
    def write_file(self, path: str, content: str) -> str
        """Write file content."""
    
    def fast_filter(self, query: str, top_n: int = 5) -> list[str]
        """Fast keyword-based filtering."""
    
    def populate_index_from_files(
        self,
        user_id: str,
        categories: list[str],
        llm_client: LocalLLMClient,
    )
        """Build index from files."""

Index Structure

{
  "index_version": 3,
  "last_updated": 1686234567.123,
  "files": [
    {
      "filename": "user1_notes.md",
      "title": "Personal Notes",
      "description": "Collection of personal notes...",
      "keywords": ["notes", "personal", "reminders"]
    }
  ]
}

tools/executor.py - Tool Execution

ToolExecutor.validate()

def validate(
    tool: Tool,
    args: dict,
) -> tuple[bool, str | None]
    """Validate tool arguments against schema.
    
    Args:
        tool: Tool definition
        args: Arguments to validate
        
    Returns:
        (is_valid, error_message)
    """

ToolExecutor.execute()

async def execute(
    self,
    name: str,
    args: dict,
) -> str
    """Execute tool with validation and timeout.
    
    1. Lookup tool by name
    2. Validate args against schema
    3. Execute tool.func(**args) with 30s timeout
    4. Wrap exceptions in safe messages
    5. Return result string
    """

Error Wrapping

try:
    result = await asyncio.wait_for(tool.func(**args), timeout=30.0)
except asyncio.TimeoutError:
    result = f"Error: tool '{name}' timed out after 30s"
except Exception as exc:
    log_event("ERROR_STACKTRACE", f"tool={name} error={exc}")
    result = f"Error: tool '{name}' failed during execution"

config/loader.py - Configuration

Config

@dataclass
class Config:
    llama: LlamaConfig
    workspace: str | None  # optional; skills storage only
    agents: dict[str, AgentConfig]
    app: AppConfig
    memory: MemoryConfig
    ensemble: EnsembleConfig
    skills: SkillsConfig
    mcp: McpConfig  # servers loaded from mcp.json only

Capability tools come from mcp.json — see docs/mcp.md. Native tools:
report_inability, load_message, load_tool_result, search_tool_results.

McpConfig

@dataclass
class McpConfig:
    enabled: bool = False
    servers: dict[str, McpServerConfig] = field(default_factory=dict)  # from mcp.json
    external_path: str | None = None

LlamaConfig

@dataclass
class LlamaConfig:
    binary_path: str
    models: dict[str, LlamaModelConfig]
    os_reserve_mb: int = 4096
    disk_reserve_mb: int = 4096

LlamaModelConfig

@dataclass
class LlamaModelConfig:
    path: str
    port: int = 8080
    ctx_size: int = 4096
    gpu_layers: int = 32
    n_batch: int = 1024
    n_ubatch: int = 256
    threads: int = 4
    threads_batch: int = 4
    flash_attn: str = "auto"
    cache_type_k: str = "f16"
    cache_type_v: str = "f16"
    optimization_mode: str = "balanced"

AgentConfig

@dataclass
class AgentConfig:
    tools: list[str]  # explicit native names; usually []; mcp_* auto-granted
    prompt_files: list[str] = field(default_factory=list)
    model: str | None = None  # llama.models key when ensemble.enabled

prompt_files: host-owned Jinja markdown stack. Empty ⇒ mode-aware bundled
system.md / system_native.md. Listed paths must exist at create_app().

AppConfig

@dataclass
class AppConfig:
    history_limit: int = 12
    max_steps: int = 20
    context_token_budget: int = 6000
    max_parse_errors: int = 5
    tool_result_inline_limit: int = 6144
    tool_result_summary_max_chars: int = 1000

See AGENTS.md and config.example.yaml for the full field set.

load_config()

def load_config(config_path: str | None = None) -> Config:
    """Load configuration from YAML file.

    Args:
        config_path: Path to config.yaml (default: "config.yaml")

    Returns:
        Config dataclass
    """

EnsembleConfig / SkillsConfig / MemoryConfig

See AGENTS.md and config.example.yaml for current fields.