Internals¶
๐๏ธ Three-Layer Agent Pattern¶
Every agent follows the same internal structure. No exceptions.
flowchart TD
MAIN["<b>__main__.py</b><br/>Server Entry<br/><i>AgentCard ยท A2AStarletteApplication ยท uvicorn</i>"]
EXEC["<b>agent_executor.py</b><br/>A2A Bridge<br/><i>AgentExecutor subclass ยท OTEL spans ยท event queue</i>"]
AGENT["<b>agent.py</b><br/>Domain Logic<br/><i>Pure business logic ยท no A2A types ยท testable in isolation</i>"]
MAIN --> EXEC
EXEC --> AGENT
| Layer | File | Responsibility |
|---|---|---|
| Server entry | __main__.py |
Defines the AgentCard, creates the A2AStarletteApplication, starts uvicorn. |
| A2A bridge | agent_executor.py |
Subclasses AgentExecutor from a2a-sdk. Translates between A2A events and domain logic. Creates OTEL spans. |
| Domain logic | agent.py |
Pure business logic. No A2A types. Async functions and generators. Testable in isolation. |
Why this matters:
agent.pynever importsa2a.*โ you can test all business logic with simple function calls and mocked LLM responsesagent_executor.pyis the only file that touches A2A types โ protocol changes are contained to one layer__main__.pyowns configuration โ if you need to change ports, skills, or card metadata, it's all in one place
Example: Mneme's Three Layers¶
async def generate_commit_messages(git_output: str, context_id: str | None = None) -> str:
messages = [
{"role": "system", "content": get_enriched_system_prompt(SYSTEM_PROMPT, "mneme")},
{"role": "user", "content": f"Generate commit message groups for these changes:\n\n```\n{git_output}\n```"},
]
return await chat("mneme", messages, temperature=0.2, max_tokens=2048, context_id=context_id) # (1)!
chat()fromkourai_common.llmโ handles model selection, timeouts, and retries automatically.
class MnemeAgentExecutor(BaseAgentExecutor): # (1)!
@executor_error_handler(agent_name="mneme")
async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
await super().execute(context, event_queue)
async def execute_agent_logic(self, context, task, updater) -> None:
user_input = context.get_user_input()
await send_working_status(updater, task, "Analyzing git changes...", emoji="๐")
full_response = ""
async for chunk in generate_commit_messages_stream(user_input):
full_response += chunk
await updater.add_artifact([Part(root=TextPart(text=full_response))])
await updater.complete()
BaseAgentExecutorfromkourai_common.base_executorโ handles task creation, input validation, and common setup. Subclasses overrideexecute_agent_logic().
agent_card = AgentCard(
name="Mneme โ Scribe",
description="Commit message specialist. Analyzes git diffs and generates grouped commit messages.",
url=get_agent_url(AGENT_NAME),
version="0.1.0",
skills=[skill],
capabilities=AgentCapabilities(streaming=True),
...
)
request_handler = DefaultRequestHandler(
agent_executor=MnemeAgentExecutor(),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(agent_card=agent_card, http_handler=request_handler)
๐ฆ Shared Library: kourai-common¶
All agents depend on a shared workspace member at shared/src/kourai_common/. The core modules are:
config.py โ Agent Configuration¶
Centralized model assignments, ports, timeouts, and environment variable handling.
from kourai_common.config import get_model, get_agent_url, MAX_ITERATIONS
get_model("metis") # (1)!
get_agent_url("techne") # (2)!
- Returns model based on
KOURAI_MODEL_TIERโ e.g."anthropic/claude-opus-4-6"on smart, Haiku on cheap (default) - Returns
"http://techne:10002/"โ agents resolve each other via Docker service names
llm.py โ Model-Agnostic LLM Interface¶
Wraps LiteLLM for async-compatible calls with per-agent timeout enforcement.
from kourai_common.llm import chat, chat_stream
# Synchronous response
result = await chat("mneme", messages, temperature=0.3, max_tokens=4096)
# Streaming response
async for chunk in chat_stream("techne", messages):
yield chunk
tracing.py โ OpenTelemetry Integration¶
Sets up distributed tracing with Jaeger as the backend.
from kourai_common.tracing import setup_tracing, create_span, get_trace_context
# Call once at startup
setup_tracing("mneme", otlp_endpoint)
# Create spans around operations
with create_span("mneme.generate", {"input_length": str(len(text))}):
result = await generate_commit_messages(text)
# Propagate trace context across agent boundaries
metadata = get_trace_context() # โ W3C traceparent/tracestate headers
retry.py โ Exponential Backoff¶
Decorator for transient failure recovery on network calls.
from kourai_common.retry import with_retry
@with_retry(max_attempts=3, base_delay=1.0,
retryable_exceptions=(httpx.ConnectError, httpx.TimeoutException))
async def send(self, text, context_id):
...
log.py โ Structured Logging¶
Configures per-agent console + rotating file logging (logs/<name>.log).
from kourai_common.log import setup_logging
log = setup_logging("mneme") # โ console + logs/mneme.log (5MB, 3 rotations)
base_executor.py โ Agent Executor Base Class¶
Common base for all agent executors. Handles task creation, input validation, and the execute โ execute_agent_logic dispatch pattern. Subclasses override execute_agent_logic().
from kourai_common.base_executor import BaseAgentExecutor
class MnemeAgentExecutor(BaseAgentExecutor):
async def execute_agent_logic(self, context, task, updater) -> None:
... # agent-specific logic
prompts.py โ System Prompt Builder¶
Constructs structured system prompts with shared personality directives and Python style standards.
from kourai_common.prompts import build_system_prompt
SYSTEM_PROMPT = build_system_prompt(
agent_name="Mneme",
role="commit message specialist",
personality="...",
specific_instructions="...",
)
messaging.py โ A2A Status Helpers¶
Convenience functions for emitting A2A status updates with consistent formatting.
from kourai_common.messaging import send_working_status
await send_working_status(updater, task, "Analyzing git changes...", emoji="๐")
memory.py โ SQLite Conversational Memory¶
Implements the persistence layer for conversational memory (see Conversational Memory below).
๐ง Conversational Memory¶
Kourai Khryseai persists the entire conversation history to a local SQLite database for context, debugging, and A2A state management. It follows a privacy-first, local-only approach.
Location: .cache/agent_memory.db
The database implements A2A Memory (Hierarchical State Management) with two primary tables:
messages: Episodic/working memory. Stores every single message exchanged, tracking thecontext_id(thread),agent_name,role, and the rawcontent.agent_states: Semantic memory. Stores structured state objects (goal hierarchies, checkpoints, summaries) for each agent and thread.