Claude Code is the most powerful agent so far. Dive into its architect!
First, Know Your Tools (commands.py, tools.py). You don’t build the tools from scratch every time you open the toolbox — they’re just sitting there, ready. In this codebase, that toolbox is a JSON file:
reference_data/commands_snapshot.json ← list of ~150 commands
reference_data/tools_snapshot.json ← list of ~100 tools
When the program starts, it reads that file once and keeps it in memory:
@lru_cache(maxsize=1)
def load_command_snapshot():
return json.loads(SNAPSHOT_PATH.read_text())Next, Understand What the User Wants (runtime.py — route_prompt). If the user says: “fix the git bug”, then the agent looks for matching words. It’s surprising how Claude Code tries to save token by using simple searching and scoring to route algorithm. It’s purely an orchestration layer.
user said: “fix”, “the”, “git”, “bug”
tool names: “git-commit”, “bash”, “file-editor”, “git-status”
“git-commit” has the word “git” in it — score 1.
“git-status” also has “git” — score 1.
“bash” has nothing — score 0.
Pick the highest scoring ones. Done. That’s the entire routing algorithm. Concretely there are functions for example
def bootstrap_session(self, prompt: str, limit: int = 5) -> RuntimeSession:
context = build_port_context()
setup_report = run_setup(trusted=True)
setup = setup_report.setup
history = HistoryLog()
engine = QueryEnginePort.from_workspace()
history.add('context', f'python_files={context.python_file_count}, archive_available={context.archive_available}')
history.add('registry', f'commands={len(PORTED_COMMANDS)}, tools={len(PORTED_TOOLS)}')
matches = self.route_prompt(prompt, limit=limit)
registry = build_execution_registry()
command_execs = tuple(registry.command(match.name).execute(prompt) for match in matches if match.kind == 'command' and registry.command(match.name))
tool_execs = tuple(registry.tool(match.name).execute(prompt) for match in matches if match.kind == 'tool' and registry.tool(match.name))
denials = tuple(self._infer_permission_denials(matches))
stream_events = tuple(engine.stream_submit_message(
prompt,
matched_commands=tuple(match.name for match in matches if match.kind == 'command'),
matched_tools=tuple(match.name for match in matches if match.kind == 'tool'),
denied_tools=denials,
))
turn_result = engine.submit_message(
prompt,
matched_commands=tuple(match.name for match in matches if match.kind == 'command'),
matched_tools=tuple(match.name for match in matches if match.kind == 'tool'),
denied_tools=denials,
)
persisted_session_path = engine.persist_session()
history.add('routing', f'matches={len(matches)} for prompt={prompt!r}')
history.add('execution', f'command_execs={len(command_execs)} tool_execs={len(tool_execs)}')
history.add('turn', f'commands={len(turn_result.matched_commands)} tools={len(turn_result.matched_tools)} denials={len(turn_result.permission_denials)} stop={turn_result.stop_reason}')
history.add('session_store', persisted_session_path)
return RuntimeSession(
prompt=prompt,
context=context,
setup=setup,
setup_report=setup_report,
system_init_message=build_system_init_message(trusted=True),
history=history,
routed_matches=matches,
turn_result=turn_result,
command_execution_messages=command_execs,
tool_execution_messages=tool_execs,
stream_events=stream_events,
persisted_session_path=persisted_session_path,
)And run_turn_loop() — “Multi-turn conversation”, A simpler loop that sends the prompt to the query engine up to max_turns times. On the first turn it uses the raw prompt; on subsequent turns it appends [turn N]. It stops early if the engine says it’s done (stop_reason != 'completed'). This mimics an agentic loop where the AI can take multiple steps.
And _infer_permission_denials() — “Is this tool safe?” A simple safety gate: any tool with “bash” in its name gets flagged as denied. This is a placeholder for a more sophisticated permission system.
def _infer_permission_denials(self, matches: list[RoutedMatch]) -> list[PermissionDenial]:
denials: list[PermissionDenial] = []
for match in matches:
if match.kind == 'tool' and 'bash' in match.name.lower():
denials.append(PermissionDenial(tool_name=match.name, reason='destructive shell execution remains gated in the Python port'))
return denialsAnd scoring engine
@staticmethod
def _score(tokens: set[str], module: PortingModule) -> int:
haystacks = [module.name.lower(), module.source_hint.lower(), module.responsibility.lower()]
score = 0
for token in tokens:
if any(token in haystack for haystack in haystacks):
score += 1
Third, Check Safety Before Doing Anything (permissions.py).
# Frozen set — immutablefs = frozenset({"bash", "rm"})fs.add("sudo")@dataclass(frozen=True)class ToolPermissionContext: deny_names: frozenset[str] = field(default_factory=frozenset) deny_prefixes: tuple[str, ...] = () @classmethod def from_iterables(cls, deny_names: list[str] | None = None, deny_prefixes: list[str] | None = None) -> 'ToolPermissionContext': return cls( deny_names=frozenset(name.lower() for name in (deny_names or [])), deny_prefixes=tuple(prefix.lower() for prefix in (deny_prefixes or [])), ) def blocks(self, tool_name: str) -> bool: lowered = tool_name.lower() return lowered in self.deny_names or any(lowered.startswith(prefix) for prefix in self.deny_prefixes)Four, Remember What Happened (transcript.py, session_store.py). Tthere are two kinds of memory and they serve different purposes.
Memory Type 1: Short-term (like RAM), it holds the recent conversation. But it has a limit. After 10 turns, it throws away the oldest ones. Why? Because AI models can only read so much text at once — they have a
context window.
@dataclassclass TranscriptStore: entries: list[str] = field(default_factory=list) flushed: bool = False def append(self, entry: str) -> None: self.entries.append(entry) self.flushed = False def compact(self, keep_last: int = 10) -> None: if len(self.entries) > keep_last: self.entries[:] = self.entries[-keep_last:] def replay(self) -> tuple[str, ...]: return tuple(self.entries) def flush(self) -> None: self.flushed = TrueMemory Type 2: Long-term (like a hard drive), This gets saved to a .json file on disk. Even if you close the program and reopen it tomorrow, you can reload this session and continue where you left off.
dataclass(frozen=True)class StoredSession: session_id: str messages: tuple[str, ...] input_tokens: int output_tokens: intDEFAULT_SESSION_DIR = Path('.port_sessions')def save_session(session: StoredSession, directory: Path | None = None) -> Path: target_dir = directory or DEFAULT_SESSION_DIR target_dir.mkdir(parents=True, exist_ok=True) path = target_dir / f'{session.session_id}.json' path.write_text(json.dumps(asdict(session), indent=2)) return pathdef load_session(session_id: str, directory: Path | None = None) -> StoredSession: target_dir = directory or DEFAULT_SESSION_DIR data = json.loads((target_dir / f'{session_id}.json').read_text()) return StoredSession( session_id=data['session_id'], messages=tuple(data['messages']), input_tokens=data['input_tokens'], output_tokens=data['output_tokens'], )Fifth, Process a Turn (query_engine.py).
from __future__ import annotationsimport jsonfrom dataclasses import dataclass, fieldfrom uuid import uuid4from .commands import build_command_backlogfrom .models import PermissionDenial, UsageSummaryfrom .port_manifest import PortManifest, build_port_manifestfrom .session_store import StoredSession, load_session, save_sessionfrom .tools import build_tool_backlogfrom .transcript import TranscriptStore@dataclass(frozen=True)class QueryEngineConfig: max_turns: int = 8 max_budget_tokens: int = 2000 compact_after_turns: int = 12 structured_output: bool = False structured_retry_limit: int = 2@dataclass(frozen=True)class TurnResult: prompt: str output: str matched_commands: tuple[str, ...] matched_tools: tuple[str, ...] permission_denials: tuple[PermissionDenial, ...] usage: UsageSummary stop_reason: str@dataclassclass QueryEnginePort: manifest: PortManifest config: QueryEngineConfig = field(default_factory=QueryEngineConfig) session_id: str = field(default_factory=lambda: uuid4().hex) mutable_messages: list[str] = field(default_factory=list) permission_denials: list[PermissionDenial] = field(default_factory=list) total_usage: UsageSummary = field(default_factory=UsageSummary) transcript_store: TranscriptStore = field(default_factory=TranscriptStore) @classmethod def from_workspace(cls) -> 'QueryEnginePort': return cls(manifest=build_port_manifest()) @classmethod def from_saved_session(cls, session_id: str) -> 'QueryEnginePort': stored = load_session(session_id) transcript = TranscriptStore(entries=list(stored.messages), flushed=True) return cls( manifest=build_port_manifest(), session_id=stored.session_id, mutable_messages=list(stored.messages), total_usage=UsageSummary(stored.input_tokens, stored.output_tokens), transcript_store=transcript, ) def submit_message( self, prompt: str, matched_commands: tuple[str, ...] = (), matched_tools: tuple[str, ...] = (), denied_tools: tuple[PermissionDenial, ...] = (), ) -> TurnResult: if len(self.mutable_messages) >= self.config.max_turns: output = f'Max turns reached before processing prompt: {prompt}' return TurnResult( prompt=prompt, output=output, matched_commands=matched_commands, matched_tools=matched_tools, permission_denials=denied_tools, usage=self.total_usage, stop_reason='max_turns_reached', ) summary_lines = [ f'Prompt: {prompt}', f'Matched commands: {", ".join(matched_commands) if matched_commands else "none"}', f'Matched tools: {", ".join(matched_tools) if matched_tools else "none"}', f'Permission denials: {len(denied_tools)}', ] output = self._format_output(summary_lines) projected_usage = self.total_usage.add_turn(prompt, output) stop_reason = 'completed' if projected_usage.input_tokens + projected_usage.output_tokens > self.config.max_budget_tokens: stop_reason = 'max_budget_reached' self.mutable_messages.append(prompt) self.transcript_store.append(prompt) self.permission_denials.extend(denied_tools) self.total_usage = projected_usage self.compact_messages_if_needed() return TurnResult( prompt=prompt, output=output, matched_commands=matched_commands, matched_tools=matched_tools, permission_denials=denied_tools, usage=self.total_usage, stop_reason=stop_reason, ) def stream_submit_message( self, prompt: str, matched_commands: tuple[str, ...] = (), matched_tools: tuple[str, ...] = (), denied_tools: tuple[PermissionDenial, ...] = (), ): yield {'type': 'message_start', 'session_id': self.session_id, 'prompt': prompt} if matched_commands: yield {'type': 'command_match', 'commands': matched_commands} if matched_tools: yield {'type': 'tool_match', 'tools': matched_tools} if denied_tools: yield {'type': 'permission_denial', 'denials': [denial.tool_name for denial in denied_tools]} result = self.submit_message(prompt, matched_commands, matched_tools, denied_tools) yield {'type': 'message_delta', 'text': result.output} yield { 'type': 'message_stop', 'usage': {'input_tokens': result.usage.input_tokens, 'output_tokens': result.usage.output_tokens}, 'stop_reason': result.stop_reason, 'transcript_size': len(self.transcript_store.entries), } def compact_messages_if_needed(self) -> None: if len(self.mutable_messages) > self.config.compact_after_turns: self.mutable_messages[:] = self.mutable_messages[-self.config.compact_after_turns :] self.transcript_store.compact(self.config.compact_after_turns) def replay_user_messages(self) -> tuple[str, ...]: return self.transcript_store.replay() def flush_transcript(self) -> None: self.transcript_store.flush() def persist_session(self) -> str: self.flush_transcript() path = save_session( StoredSession( session_id=self.session_id, messages=tuple(self.mutable_messages), input_tokens=self.total_usage.input_tokens, output_tokens=self.total_usage.output_tokens, ) ) return str(path) def _format_output(self, summary_lines: list[str]) -> str: if self.config.structured_output: payload = { 'summary': summary_lines, 'session_id': self.session_id, } return self._render_structured_output(payload) return '\n'.join(summary_lines) def _render_structured_output(self, payload: dict[str, object]) -> str: last_error: Exception | None = None for _ in range(self.config.structured_retry_limit): try: return json.dumps(payload, indent=2) except (TypeError, ValueError) as exc: # pragma: no cover - defensive branch last_error = exc payload = {'summary': ['structured output retry'], 'session_id': self.session_id} raise RuntimeError('structured output rendering failed') from last_error def render_summary(self) -> str: command_backlog = build_command_backlog() tool_backlog = build_tool_backlog() sections = [ '# Python Porting Workspace Summary', '', self.manifest.to_markdown(), '', f'Command surface: {len(command_backlog.modules)} mirrored entries', *command_backlog.summary_lines()[:10], '', f'Tool surface: {len(tool_backlog.modules)} mirrored entries', *tool_backlog.summary_lines()[:10], '', f'Session id: {self.session_id}', f'Conversation turns stored: {len(self.mutable_messages)}', f'Permission denials tracked: {len(self.permission_denials)}', f'Usage totals: in={self.total_usage.input_tokens} out={self.total_usage.output_tokens}', f'Max turns: {self.config.max_turns}', f'Max budget tokens: {self.config.max_budget_tokens}', f'Transcript flushed: {self.transcript_store.flushed}', ] return '\n'.join(sections)Sixth, Start Up in the Right Order (bootstrap_graph.py), This codebase defines a startup sequence explicitly:
Stage 1: Prefetch stuff in the background
Stage 2: Check the environment is safe
Stage 3: Parse what the user typed
Stage 4: Load commands and tools
Stage 5: ← TRUST GATE ← load plugins only if trusted
Stage 6: Pick a mode (local / remote / ssh)
Stage 7: Start the conversation loop
@dataclass(frozen=True)class BootstrapGraph: stages: tuple[str, ...] def as_markdown(self) -> str: lines = ['# Bootstrap Graph', ''] lines.extend(f'- {stage}' for stage in self.stages) return '\n'.join(lines)def build_bootstrap_graph() -> BootstrapGraph: return BootstrapGraph( stages=( 'top-level prefetch side effects', 'warning handler and environment guards', 'CLI parser and pre-action trust gate', 'setup() + commands/agents parallel load', 'deferred init after trust', 'mode routing: local / remote / ssh / teleport / direct-connect / deep-link', 'query engine submit loop', ) )It’s documentation as code. Instead of writing the startup order in a wiki that gets stale, they encoded it as a data structure that the CLI can print (main.py’s bootstrap-graph subcommand). This way the documentation stays in the repo and is always accessible via: