Class MyAgent

mcp_demo.py:94–168

Assistant agent with support for both function-based and MCP tools

class MyAgent(BaseAgent):
    """Assistant agent with support for both function-based and MCP tools"""
    
    def __init__(self, model: BaseModel, config: Dict[str, Any]):
        super().__init__(model, config)
        
        # Get tool registry
        self.tool_registry = get_registry()
        
        # Get allowed tools list from config (whitelist mode)
        self.allowed_tools = config.get("allowed_tools", None)
        
        logger.info(f"MyAgent '{self.name}' initialized")
        if self.allowed_tools:
            logger.info(f"Allowed tools: {self.allowed_tools}")
        else:
            logger.info(f"All tools available")
    
    async def _execute_tool(self, function_name: str, function_args: Dict[str, Any]) -> Any:
        """Execute tool function (routes to either function-based or MCP tool)"""
        logger.info(f"Executing tool: {function_name}")
        logger.debug(f"Arguments: {json.dumps(function_args, ensure_ascii=False)}")
        
        try:
            result = await self.tool_registry.execute(function_name, **function_args)
            logger.info(f"Tool execution successful: {function_name}")
            logger.info(f"Tool Call Result: {result}")
            return result
        except Exception as e:
            logger.error(f"Tool execution failed for {function_name}: {e}")
            raise
    
    async def get_registered_tools(self) -> list:
        """Get all tool definitions in OpenAI format (with permission filtering applied)"""
        return await self.tool_registry.get_all_definitions(
            allowed_tools=self.allowed_tools
        )
    
    async def execute(self, context: Dict[str, Any], params: Dict[str, Any]) -> Dict[str, Any]:
        """Execute agent task"""
        query = context.get("query", "")
        
        if not query:
            return {"status": "error", "error": "No query provided"}
        
        logger.info(f"Processing query: {query}")
        
        try:
            tools = await self.get_registered_tools()
            logger.info(f"Available tools: {len(tools)}")


            # Retrieve related tools
            related_tools = get_related_tools(query, tools)
            logger.info(f"Related tools found: {len(related_tools)}")
                
            response = await self._call_model_with_tools(
                system_prompt="",