Agent Claude¶

agent_claude — mvp.agent_claude

Cluster: Agents & LLM | Type: component | MCP Tools: 12

Overview¶

Beta, review-pending Anthropic Claude agent with tool calling, autonomous tool-use loops, retry policy, circuit breaker, and lifecycle management. Block contract verification_method=tier1_review_pending; @component_maturity("beta").

When to use:

Single-shot Claude inference with tool calling (infer())
Autonomous multi-step agent execution: call → observe → execute → repeat (infer_loop())
Production deployments requiring retry, circuit breaker, and structured logging

Example (single-shot):

from mvp.agent_claude import AgentClaudeBlock, AgentInput, MessageDict

block = AgentClaudeBlock(name="claude")
result = block.infer(AgentInput(
    messages=[MessageDict(role="user", content="Explain monads.")],
    allow_paid_api=True,
))
# result.ok → True; result.value → AgentOutput with response text and usage info

Example (autonomous tool-use loop):

from mvp.agent_claude import AgentClaudeBlock, AgentLoopInput, MessageDict, ToolSpec

def tool_executor(tool_name: str, tool_input: dict) -> str:
    if tool_name == "calculator":
        return str(eval(tool_input["expression"]))
    return f"Unknown tool: {tool_name}"

block = AgentClaudeBlock(name="claude")
result = block.infer_loop(
    AgentLoopInput(
        messages=[MessageDict(role="user", content="What is 6 * 7 + 3?")],
        tools=[ToolSpec(name="calculator", description="Evaluate math", input_schema={"type": "object", "properties": {"expression": {"type": "string"}}})],
        max_iterations=5,
        allow_paid_api=True,
    ),
    tool_executor=tool_executor,
)
# result.value.iterations, result.value.tool_calls_made, result.value.response

Works well with: llm_router, goal_engine, align_csf

Residual Production Risks¶

The component is functional and hardened for pilot use, but these risks should remain visible during launch readiness checks:

Risk	Production impact	Current mitigation
Clean-machine MCP install is not fully proven until manually tested end-to-end	A new user may fail before reaching their first useful workflow, which directly conflicts with the 10-minute onboarding target	Run a fresh-machine install smoke test before each pilot cohort: install G6, add the MCP server, call `claude_status`, then complete one starter workflow
Claude Code CLI path and authentication depend on the user's local environment	`agent_claude` MCP tools can report degraded status or fail if `claude` is not on `PATH` or the user is not authenticated	`claude_status` reports CLI availability and key configuration; setup docs should tell users to verify Claude Code works before adding G6
Model-tier recommendation is a caller-cost advisory cap, not a per-request execution budget predicate	A direct execution caller can choose any supported model ID unless an outer budget system constrains the call	`claude_code_run` rejects unsupported model IDs before subprocess execution and `recommend_model` reports `execution_model_allowlist_enforced=True`, but central spend caps remain external
Paid API calls are consent-gated but not a complete quota or billing-enforcement system	A local stdio user can opt into paid Anthropic/OpenRouter calls without a central spending cap enforced by this component	`allow_paid_api=True` or `ALLOW_PAID_API=1` is required before SDK calls; use platform-level billing, rate limits, and external quotas for production cost control

Public API¶

`AgentClaudeBlock(LifecycleMixin, AIBlock[AgentInput, AgentOutput, None])`¶

Beta, review-pending Claude agent with retry, circuit breaker, and lifecycle.

Field	Type	Default
`name`	`str`	'agent_claude'
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)
`api_key`	`str`	''
`timeout`	`float`	30.0
`retry_policy`	`RetryPolicy`	field(default_factory=lambda: RetryPolicy(max_attempts=3, initial_delay=1.0, jitter=True, retryable_exceptions=(ConnectionError, TimeoutError, OSError), retry_on_transient_provider=True))

Methods:

`capabilities() -> dict[str, Any]`¶

Return zero-cost Tier 1 source health without exposing secret values.

`infer(data: AgentInput) -> Result[AgentOutput]`¶

`infer_loop(data: AgentLoopInput, tool_executor: ToolExecutor) -> Result[AgentLoopOutput]`¶

Autonomous tool-use loop: call → observe → execute tools → repeat.

`MessageDict(BaseModel)`¶

A single conversation message.

Field	Type	Default
`role`	`Literal['user', 'assistant']`	required
`content`	`str`	required

`ToolSpec(BaseModel)`¶

Specification for a tool the agent may call.

Field	Type	Default
`name`	`str`	required
`description`	`str`	required
`input_schema`	`dict[str, Any]`	Field(default_factory=dict)

`ToolCall(BaseModel)`¶

A tool call made by the agent.

Field	Type	Default
`tool_name`	`str`	required
`tool_input`	`dict[str, Any]`	Field(default_factory=dict)
`tool_use_id`	`str`	''

`AgentInput(BaseModel)`¶

Input to AgentClaudeBlock.

Field	Type	Default
`messages`	`list[MessageDict]`	required
`system_prompt`	`str`	'You are a helpful AI assistant.'
`model`	`str`	'claude-haiku-4-5-20251001'
`tools`	`list[ToolSpec]`	Field(default_factory=list)
`max_tokens`	`int`	1024
`allow_paid_api`	`bool`	False

`UsageInfo(BaseModel)`¶

Token usage statistics.

Field	Type	Default
`input_tokens`	`int`	0
`output_tokens`	`int`	0

`ToolResult(BaseModel)`¶

Result from executing a tool.

Field	Type	Default
`tool_use_id`	`str`	required
`content`	`str`	required
`is_error`	`bool`	False
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None

`ToolExecutor(Protocol)`¶

Protocol for executing tool calls. Implementations dispatch by tool_name.

`AgentLoopInput(BaseModel)`¶

Input for autonomous tool-use loop execution.

Field	Type	Default
`messages`	`list[MessageDict]`	required
`system_prompt`	`str`	'You are a helpful AI assistant.'
`model`	`str`	'claude-haiku-4-5-20251001'
`tools`	`list[ToolSpec]`	Field(default_factory=list)
`max_tokens`	`int`	1024
`allow_paid_api`	`bool`	False
`max_iterations`	`int`	10

`AgentLoopOutput(BaseModel)`¶

Output from autonomous tool-use loop execution.

Field	Type	Default
`response`	`str`	required
`iterations`	`int`	1
`tool_calls_made`	`list[ToolCall]`	Field(default_factory=list)
`tool_results`	`list[ToolResult]`	Field(default_factory=list)
`usage`	`UsageInfo`	Field(default_factory=UsageInfo)
`model`	`str`	''
`stop_reason`	`str`	'end_turn'
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`Literal['verified', 'qualified-draft', 'blocked-escalated']`	'qualified-draft'
`warning_card`	`dict[str, Any]`	Field(default_factory=dict)
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''

`AgentOutput(BaseModel)`¶

Output from AgentClaudeBlock.

Field	Type	Default
`response`	`str`	required
`tool_calls`	`list[ToolCall]`	Field(default_factory=list)
`usage`	`UsageInfo`	Field(default_factory=UsageInfo)
`model`	`str`	required
`stop_reason`	`str`	'end_turn'
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`Literal['verified', 'qualified-draft', 'blocked-escalated']`	'qualified-draft'
`warning_card`	`dict[str, Any]`	Field(default_factory=dict)
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''

`AgentClaudeMCPBlock(AIBlock['MCPClaudeInput', 'MCPClaudeOutput', dict])`¶

Deep 7 execution ops + 2 advisory ops (recommend_model, list_patterns) = 9 MCP ops.

Field	Type	Default
`name`	`str`	'agent_claude_mcp'
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)
`db_path`	`str`	''
`agentic_planner`	`object`	None

Methods:

`infer(inp: 'MCPClaudeInput') -> 'Result[MCPClaudeOutput]'`¶

`MCPClaudeInput(BaseModel)`¶

Field	Type	Default
`op`	`Literal['run', 'session', 'skill', 'agent', 'meta', 'pipeline', 'status', 'source_capabilities', 'recommend_model', 'list_patterns', 'ops', 'help']`	required
`prompt`	`str`	''
`tools_json`	`str`	''
`session_id`	`str`	''
`model`	`str`	'claude-haiku-4-5-20251001'
`max_tokens`	`int`	1024
`max_cost_tier`	`str`	''
`system_prompt`	`str`	''
`sub_op`	`str`	''
`messages_json`	`str`	''
`metadata_json`	`str`	''
`name`	`str`	''
`content`	`str`	''
`query`	`str`	''
`config_json`	`str`	''
`target`	`str`	''
`scope`	`str`	''
`tasks_json`	`str`	''
`tags`	`list[str]`	Field(default_factory=list)
`limit`	`int`	20

`MCPClaudeOutput(BaseModel)`¶

Field	Type	Default
`op`	`str`	required
`success`	`bool`	required
`data`	`dict`	Field(default_factory=dict)
`message`	`str`	''
`error`	`str`	''
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`Literal['verified', 'qualified-draft', 'blocked-escalated']`	'qualified-draft'
`warning_card`	`dict[str, Any]`	Field(default_factory=dict)
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''

`SessionRecord(BaseModel)`¶

Field	Type	Default
`id`	`str`	required
`title`	`str`	''
`messages`	`list[dict]`	Field(default_factory=list)
`metadata`	`dict`	Field(default_factory=dict)
`created_at`	`str`	''
`updated_at`	`str`	''

`AgentRecord(BaseModel)`¶

Field	Type	Default
`id`	`str`	required
`name`	`str`	required
`role`	`str`	''
`tools`	`list[dict]`	Field(default_factory=list)
`config`	`dict`	Field(default_factory=dict)
`created_at`	`str`	''

MCP Tools¶

Operation	Source
`run`	claude_mcp
`session`	claude_mcp
`skill`	claude_mcp
`agent`	claude_mcp
`meta`	claude_mcp
`pipeline`	claude_mcp
`status`	claude_mcp
`source_capabilities`	claude_mcp
`recommend_model`	claude_mcp
`list_patterns`	claude_mcp
`ops`	claude_mcp
`help`	claude_mcp

Agent Claude¶

Overview¶

Residual Production Risks¶

Public API¶

AgentClaudeBlock(LifecycleMixin, AIBlock[AgentInput, AgentOutput, None])¶

capabilities() -> dict[str, Any]¶

infer(data: AgentInput) -> Result[AgentOutput]¶

infer_loop(data: AgentLoopInput, tool_executor: ToolExecutor) -> Result[AgentLoopOutput]¶

MessageDict(BaseModel)¶

ToolSpec(BaseModel)¶

ToolCall(BaseModel)¶

AgentInput(BaseModel)¶

UsageInfo(BaseModel)¶

ToolResult(BaseModel)¶

ToolExecutor(Protocol)¶

AgentLoopInput(BaseModel)¶

AgentLoopOutput(BaseModel)¶

AgentOutput(BaseModel)¶

AgentClaudeMCPBlock(AIBlock['MCPClaudeInput', 'MCPClaudeOutput', dict])¶

infer(inp: 'MCPClaudeInput') -> 'Result[MCPClaudeOutput]'¶

MCPClaudeInput(BaseModel)¶

MCPClaudeOutput(BaseModel)¶

SessionRecord(BaseModel)¶

AgentRecord(BaseModel)¶