from typing import Any
import mcp.types as types
from hud.agents import MCPAgent
from hud.types import AgentResponse, MCPToolCall, MCPToolResult
class MyAgent(MCPAgent):
"""Custom agent using any LLM provider."""
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.client = MyLLMClient() # Your LLM client
async def get_system_messages(self) -> list[Any]:
"""Return system prompt in your LLM's format. Called once at start."""
return [{"role": "system", "content": self.system_prompt or "You are a helpful assistant."}]
async def get_response(self, messages: list[Any]) -> AgentResponse:
"""Call your LLM and return its response. Called each turn."""
response = await self.client.chat(
messages=messages,
tools=self.get_tool_schemas(), # MCP tools formatted for your LLM
)
# Convert to AgentResponse
tool_calls = [
MCPToolCall(name=tc.name, arguments=tc.arguments)
for tc in response.tool_calls
] if response.tool_calls else []
return AgentResponse(
content=response.content,
tool_calls=tool_calls,
done=len(tool_calls) == 0, # Done when no more tool calls
)
async def format_blocks(self, blocks: list[types.ContentBlock]) -> list[Any]:
"""Convert MCP content blocks to your LLM's message format."""
content = []
for block in blocks:
if isinstance(block, types.TextContent):
content.append({"type": "text", "text": block.text})
elif isinstance(block, types.ImageContent):
content.append({"type": "image", "data": block.data})
return [{"role": "user", "content": content}]
async def format_tool_results(
self, tool_calls: list[MCPToolCall], tool_results: list[MCPToolResult]
) -> list[Any]:
"""Convert MCP tool results to your LLM's format."""
return [
{"role": "tool", "tool_call_id": call.id, "content": str(result.content)}
for call, result in zip(tool_calls, tool_results)
]