68 lines
2.2 KiB
Python
68 lines
2.2 KiB
Python
"""Tool calling via a stdio MCP server.
|
|
|
|
`BackendOptions.mcp_servers` materializes into a temp `--mcp-config` JSON
|
|
file under the hood. The model decides when to invoke the tool, claude's
|
|
TUI runs the round-trip, and the events stream surfaces a `ToolUseBlock`
|
|
plus its `ToolResultBlock`.
|
|
|
|
This example wires the bundled echo server at `scripts/echo_mcp_server.py`
|
|
— same one the smoke test uses.
|
|
|
|
Run:
|
|
python examples/mcp_tool.py
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from claude_code_api import (
|
|
AssistantMessage,
|
|
BackendOptions,
|
|
ClaudeCodeBackend,
|
|
ResultMessage,
|
|
TextBlock,
|
|
ToolResultBlock,
|
|
ToolUseBlock,
|
|
UserMessage,
|
|
)
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
ECHO_SCRIPT = REPO_ROOT / "scripts" / "echo_mcp_server.py"
|
|
CWD = Path(os.environ.get("CLAUDE_CODE_CWD", Path.cwd())).resolve()
|
|
|
|
|
|
async def main() -> None:
|
|
opts = BackendOptions(
|
|
cwd=str(CWD),
|
|
dangerously_skip_permissions=True,
|
|
mcp_servers={
|
|
"echo": {"command": sys.executable, "args": [str(ECHO_SCRIPT)]},
|
|
},
|
|
)
|
|
async with ClaudeCodeBackend(opts) as backend:
|
|
prompt = (
|
|
"Call the `mcp__echo__echo` tool with text='ping' and then "
|
|
"tell me what it returned. Reply with one short sentence."
|
|
)
|
|
async for event in backend.complete([{"role": "user", "content": prompt}]):
|
|
if isinstance(event, AssistantMessage):
|
|
for block in event.content:
|
|
if isinstance(block, TextBlock) and block.text.strip():
|
|
print(f"[assistant] {block.text.strip()}")
|
|
elif isinstance(block, ToolUseBlock):
|
|
print(f"[tool_use] {block.name}({block.input})")
|
|
elif isinstance(event, UserMessage) and isinstance(event.content, list):
|
|
for block in event.content:
|
|
if isinstance(block, ToolResultBlock):
|
|
print(f"[tool_res] {block.content!r}")
|
|
elif isinstance(event, ResultMessage):
|
|
print(f"[result] stop_reason={event.stop_reason}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|