Key Abstractions & Design Patterns

@dataclass(frozen=True)
class Runtime:
    config: Config                      # Application configuration
    oauth: OAuthManager                 # OAuth token management
    llm: LLM | None                     # LLM instance (None if not configured)
    session: Session                    # Session metadata
    builtin_args: BuiltinSystemPromptArgs  # Template variables
    denwa_renji: DenwaRenji            # D-Mail manager
    approval: Approval                  # Tool approval system
    labor_market: LaborMarket          # Subagent registry
    environment: Environment           # System environment info
    skills: Mapping[str, DiscoveredSkill]  # Available skills

@staticmethod
async def create(
    config: Config,
    oauth: OAuthManager,
    llm: LLM | None,
    session: Session,
    *,
    yolo: bool = False,
    skills_roots: Sequence[Path] = (),
) -> Runtime:
    # Creates all dependencies and returns frozen Runtime

def copy_for_fixed_subagent(self) -> Runtime:
    """Clone for fixed subagent with separate DenwaRenji."""
 
def copy_for_dynamic_subagent(self) -> Runtime:
    """Clone for dynamic subagent with shared LaborMarket."""

@dataclass(frozen=True)
class Agent:
    name: str                  # Agent identifier
    system_prompt: str         # Rendered system prompt
    toolset: KimiToolset       # Available tools
    runtime: Runtime           # Execution context

class Context:
    def __init__(self, backend: Path | None = None):
        self._messages: list[Message] = []
        self._checkpoints: dict[int, int] = {}  # id -> message index
        self._backend = backend  # File path for persistence
 
    def append_message(self, message: Message) -> None:
        """Add message and persist to backend."""
 
    def checkpoint(self, checkpoint_id: int) -> None:
        """Create a checkpoint at current position."""
 
    def revert_to(self, checkpoint_id: int) -> None:
        """Remove messages after checkpoint (time-travel)."""
 
    @classmethod
    def restore(cls, backend: Path) -> Context:
        """Load context from file backend."""

{"role": "user", "content": [...]}
{"role": "assistant", "content": [...]}
{"role": "_checkpoint", "id": 0}
{"role": "_usage", "token_count": 1500}

class KimiToolset(Toolset):
    def add(self, tool: Tool) -> None:
        """Register a tool."""
 
    def find(self, name: str) -> Tool | None:
        """Find tool by name."""
 
    async def handle(self, tool_call: ToolCall) -> ToolResult:
        """Execute tool call with context tracking."""
 
    def load_tools(
        self,
        paths: Sequence[str],
        dependencies: dict[type, Any],
    ) -> None:
        """Load tools from module paths with DI."""
 
    async def load_mcp_tools(
        self,
        mcp_configs: Sequence[MCPServerConfig],
        runtime: Runtime,
    ) -> None:
        """Load tools from MCP servers (async background)."""

class Approval:
    def __init__(self, *, yolo: bool = False):
        self._yolo = yolo
        self._auto_approved: set[tuple[str, str]] = set()
        self._pending: asyncio.Queue[Request] = asyncio.Queue()
 
    async def request(
        self,
        sender: str,
        action: str,
        description: str,
        display: str | None = None,
    ) -> bool:
        """Request approval for an action."""
        if self._yolo:
            return True
        if (sender, action) in self._auto_approved:
            return True
 
        # Create request and wait for resolution
        request = Request(sender, action, description, display)
        self._pending.put_nowait(request)
        return await request.future
 
    def fetch_request(self) -> Request | None:
        """Soul fetches pending approval requests."""
 
    def resolve_request(
        self,
        request: Request,
        response: Literal["approve", "approve_for_session", "reject"],
    ) -> None:
        """Soul resolves with user's decision."""

class DenwaRenji:
    """D-Mail manager for checkpoint-based time travel."""
 
    def send_dmail(self, dmail: DMail) -> None:
        """Tool sends D-Mail to be processed."""
 
    def set_n_checkpoints(self, n: int) -> None:
        """Track current checkpoint count."""
 
    def fetch_pending_dmail(self) -> DMail | None:
        """Soul fetches and clears pending D-Mail."""
 
@dataclass
class DMail:
    checkpoint_id: int
    messages: list[Message]

@dataclass
class LaborMarket:
    fixed_subagents: dict[str, Agent]    # Pre-loaded from agent.yaml
    dynamic_subagents: dict[str, Agent]  # Created via CreateSubagent
 
    def register_dynamic(self, name: str, agent: Agent) -> None:
        """Register a dynamically created subagent."""
 
    def find(self, name: str) -> Agent | None:
        """Find subagent by name (checks both registries)."""

# Simple tool with manual parameter handling
class MyTool(CallableTool):
    name = "my_tool"
    description = "Does something"
    params = {...}  # JSON Schema
 
    async def __call__(self, **kwargs) -> ToolReturnValue:
        ...
 
# Typed tool with Pydantic params
class MyTypedTool(CallableTool2[MyParams]):
    name = "my_typed_tool"
    description = "Does something with typed params"
 
    async def __call__(self, params: MyParams) -> ToolReturnValue:
        ...

# Success with content
return ToolOk("Operation completed")
return ToolOk(content=[TextPart("Result"), ImagePart(data)])
 
# Error
return ToolError("Something went wrong")
 
# Rejection (user denied approval)
return ToolRejectedError("User rejected the action")

class Shell(CallableTool2[ShellParams]):
    def __init__(self, approval: Approval, environment: Environment):
        self._approval = approval
        self._environment = environment

def load_tools(self, paths: Sequence[str], deps: dict[type, Any]):
    for path in paths:
        module_path, class_name = path.rsplit(":", 1)
        module = importlib.import_module(module_path)
        tool_class = getattr(module, class_name)
 
        # Inspect __init__ for dependencies
        sig = inspect.signature(tool_class.__init__)
        kwargs = {}
        for param in sig.parameters.values():
            if param.annotation in deps:
                kwargs[param.name] = deps[param.annotation]
 
        tool = tool_class(**kwargs)
        self.add(tool)

# toolset.py:355
class MCPTool(CallableTool):
    """Wrapper for MCP server tools."""
 
    def __init__(self, mcp_tool: fastmcp.Tool, client: MCPClient):
        self.name = mcp_tool.name
        self.description = mcp_tool.description
        self.params = mcp_tool.inputSchema
        self._client = client
 
    async def __call__(self, **kwargs) -> ToolReturnValue:
        result = await self._client.call_tool(self.name, kwargs)
        return ToolOk(result.content)

# Good: Dependencies injected
class WriteFile(CallableTool2[WriteFileParams]):
    def __init__(self, approval: Approval):
        self._approval = approval
 
# Usage in loader
tool = WriteFile(approval=runtime.approval)

class Soul(Protocol):
    async def run(self, user_input: str) -> AsyncIterator[WireMessage]:
        ...
 
    def compact_context(self) -> AsyncIterator[WireMessage]:
        ...

# toolset.py
current_tool_call: ContextVar[ToolCall | None] = ContextVar(
    "current_tool_call", default=None
)
 
def get_current_tool_call_or_none() -> ToolCall | None:
    return current_tool_call.get()
 
# In handle()
async def handle(self, tool_call: ToolCall):
    token = current_tool_call.set(tool_call)
    try:
        result = await tool(tool_call.arguments)
    finally:
        current_tool_call.reset(token)

# Soul sends events
yield WireMessage.MessagePart(TextPart("Hello"))
yield WireMessage.ToolCallRequest(tool_call)
yield WireMessage.ApprovalRequest(request)
 
# UI receives and renders
async for msg in soul.run(user_input):
    match msg:
        case WireMessage.MessagePart(part):
            render_part(part)
        case WireMessage.ApprovalRequest(req):
            response = prompt_user(req)
            soul.resolve_approval(req, response)

class Config(BaseModel):
    model_config = ConfigDict(frozen=True)
 
    models: dict[str, LLMModel]
    providers: dict[str, LLMProvider]
    loop_control: LoopControl

# Create checkpoint
context.checkpoint(checkpoint_id=5)
 
# Later: revert to checkpoint
context.revert_to(checkpoint_id=5)  # Removes all messages after checkpoint 5

# KimiCLI.create() - factory for entire application
kimi = await KimiCLI.create(session, config, model_name=...)
 
# Runtime.create() - factory for runtime dependencies
runtime = await Runtime.create(config, oauth, llm, session, ...)
 
# load_agent() - factory for agent with tools
agent = await load_agent(agent_file, runtime, mcp_configs)