Spaces:
Sleeping
Sleeping
| import asyncio | |
| import anthropic | |
| from mcp import ClientSession | |
| from mcp.client.sse import sse_client | |
| from .base import Agent, AgentResult, ToolCall | |
| _DEFAULT_MODEL = "claude-sonnet-4-6" | |
| _SYSTEM = ( | |
| "You are a harmonic analysis assistant. After analysing a chord sequence, present only " | |
| "well-known results — songs that are recognisable hits or by widely-known artists. " | |
| "Use your knowledge to filter out obscure tracks. If no well-known matches exist, say so." | |
| ) | |
| class ClaudeAgent(Agent): | |
| """Agent using the Anthropic SDK, connecting to tools via an MCP SSE server.""" | |
| def __init__(self, mcp_url: str, model: str = _DEFAULT_MODEL, client: anthropic.Anthropic | None = None): | |
| super().__init__(mcp_url) | |
| self.model = model | |
| self._client = client or anthropic.Anthropic() | |
| def run(self, user_prompt: str) -> AgentResult: | |
| return asyncio.run(self._run(user_prompt)) | |
| async def _run(self, user_prompt: str) -> AgentResult: | |
| async with sse_client(self.mcp_url) as (read, write): | |
| async with ClientSession(read, write) as session: | |
| await session.initialize() | |
| tools = [_to_anthropic(t) for t in (await session.list_tools()).tools] | |
| messages = [{"role": "user", "content": user_prompt}] | |
| tool_calls: list[ToolCall] = [] | |
| while True: | |
| response = self._client.messages.create( | |
| model=self.model, | |
| max_tokens=4096, | |
| system=_SYSTEM, | |
| tools=tools, | |
| messages=messages, | |
| ) | |
| if response.stop_reason == "end_turn": | |
| text = next( | |
| (b.text for b in response.content if b.type == "text"), "" | |
| ) | |
| return AgentResult(response=text, tool_calls=tool_calls) | |
| tool_results = [] | |
| for block in response.content: | |
| if block.type == "tool_use": | |
| result_text = await self._invoke_tool(session, block.name, dict(block.input)) | |
| tool_calls.append(ToolCall(name=block.name, args=dict(block.input), result=result_text)) | |
| tool_results.append({ | |
| "type": "tool_result", | |
| "tool_use_id": block.id, | |
| "content": result_text, | |
| }) | |
| messages.append({"role": "assistant", "content": response.content}) | |
| messages.append({"role": "user", "content": tool_results}) | |
| async def _invoke_tool(self, session: ClientSession, name: str, args: dict) -> str: | |
| mcp_result = await session.call_tool(name, args) | |
| return "\n".join(item.text for item in mcp_result.content if hasattr(item, "text")) | |
| def _to_anthropic(tool) -> dict: | |
| return { | |
| "name": tool.name, | |
| "description": tool.description or "", | |
| "input_schema": tool.inputSchema, | |
| } | |