AutoGen MCP Server

Instructions

Implementation Reference

• src/autogen_mcp/server.py:103-177 (handler)

  Core handler function that implements the create_agent tool logic. Parses arguments, creates specific AutoGen agent types (AssistantAgent, UserProxyAgent, ConversableAgent, etc.), registers tools if provided, adds to agent manager, and returns success info.
  async def _create_agent(self, args: Dict[str, Any]) -> Dict[str, Any]: """Create an enhanced AutoGen agent.""" name = args["name"] agent_type = args["type"] system_message = args.get("system_message", "You are a helpful AI assistant.") llm_config = args.get("llm_config", self.server_config.default_llm_config) code_execution_config = args.get("code_execution_config", self.server_config.default_code_execution_config) human_input_mode = args.get("human_input_mode", "NEVER") tools = args.get("tools", []) teachability_config = args.get("teachability", {}) try: if agent_type == "assistant": agent = AssistantAgent( name=name, system_message=system_message, llm_config=llm_config, human_input_mode=human_input_mode, ) elif agent_type == "user_proxy": agent = UserProxyAgent( name=name, system_message=system_message, code_execution_config=code_execution_config, human_input_mode=human_input_mode, ) elif agent_type == "conversable": agent = ConversableAgent( name=name, system_message=system_message, llm_config=llm_config, human_input_mode=human_input_mode, ) elif agent_type == "teachable" and TeachableAgent: agent = TeachableAgent( name=name, system_message=system_message, llm_config=llm_config, teach_config=teachability_config, ) elif agent_type == "retrievable" and RetrieveUserProxyAgent: agent = RetrieveUserProxyAgent( name=name, system_message=system_message, code_execution_config=code_execution_config, retrieve_config=args.get("retrieve_config", {}), ) else: return {"error": f"Unknown or unsupported agent type: {agent_type}"} # Register tools if provided if tools: for tool in tools: if hasattr(agent, 'register_for_execution'): agent.register_for_execution(name=tool["name"])(tool["function"]) self.agent_manager.add_agent(name, agent) return { "success": True, "message": f"Agent '{name}' created successfully", "agent_info": { "name": name, "type": agent_type, "capabilities": { "llm_enabled": hasattr(agent, "llm_config") and agent.llm_config is not None, "code_execution": hasattr(agent, "code_execution_config") and agent.code_execution_config is not False, "teachable": agent_type == "teachable", "tools_count": len(tools) } } } except Exception as e: return {"error": f"Failed to create agent: {str(e)}"}
• src/autogen_mcp/config.py:7-40 (schema)

  Dataclass defining the input schema/configuration for agent creation with fields matching the expected arguments in create_agent handler.
  class AgentConfig: """Configuration for an AutoGen agent.""" name: str type: str = "assistant" # 'assistant' or 'user' role: str = "assistant" # For compatibility description: str = "" system_message: str = "" llm_config: Optional[Dict[str, Any]] = None code_execution_config: Optional[Dict[str, Any]] = None def to_autogen_config(self) -> Dict[str, Any]: """Convert to AutoGen configuration.""" config = { "name": self.name, "human_input_mode": "NEVER", # MCP handles input "max_consecutive_auto_reply": 10, # Reasonable default "system_message": self.system_message or None, "llm_config": self.llm_config or {}, "code_execution_config": self.code_execution_config or False, } # Add type-specific settings if self.type == "assistant": config.update({ "is_termination_msg": lambda x: "TERMINATE" in x.get("content", ""), }) elif self.type == "user": config.update({ "human_input_mode": "NEVER", "code_execution_config": False, # User agents don't execute code }) return config
• src/autogen_mcp/server.py:73-102 (registration)

  Tool dispatch/registration logic in handle_tool_call method where 'create_agent' is explicitly checked and routed to the handler.
  async def handle_tool_call(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: """Handle tool calls with enhanced AutoGen features.""" try: if tool_name == "create_agent": return await self._create_agent(arguments) elif tool_name == "create_workflow": return await self._create_workflow(arguments) elif tool_name == "execute_chat": return await self._execute_chat(arguments) elif tool_name == "execute_group_chat": return await self._execute_group_chat(arguments) elif tool_name == "execute_nested_chat": return await self._execute_nested_chat(arguments) elif tool_name == "execute_swarm": return await self._execute_swarm(arguments) elif tool_name == "execute_workflow": return await self._execute_workflow(arguments) elif tool_name == "manage_agent_memory": return await self._manage_agent_memory(arguments) elif tool_name == "configure_teachability": return await self._configure_teachability(arguments) elif tool_name == "get_agent_status": return await self._get_agent_status(arguments) elif tool_name == "get_resource": return await self._get_resource(arguments) else: return {"error": f"Unknown tool: {tool_name}"} except Exception as e: return {"error": str(e)}
• src/autogen_mcp/agents.py:16-52 (helper)

  Supporting method in AgentManager for creating agents using AgentConfig, similar logic to the main handler.
  def create_agent(self, config: AgentConfig) -> ConversableAgent: """Create a new agent.""" if config.name in self._agents: raise ValueError(f"Agent {config.name} already exists") # Get base configuration agent_config = config.to_autogen_config() # Add default configurations if not provided if not agent_config.get("llm_config"): agent_config["llm_config"] = self._server_config.get_default_llm_config() if not agent_config.get("code_execution_config") and config.type == "assistant": agent_config["code_execution_config"] = self._server_config.get_default_code_execution_config() # Create the appropriate agent type if config.type == "assistant": agent = autogen.AssistantAgent( name=config.name, system_message=agent_config.get("system_message", ""), llm_config=agent_config.get("llm_config"), code_execution_config=agent_config.get("code_execution_config"), human_input_mode="NEVER", max_consecutive_auto_reply=10, ) elif config.type == "user": agent = autogen.UserProxyAgent( name=config.name, human_input_mode="NEVER", max_consecutive_auto_reply=10, system_message=agent_config.get("system_message", ""), code_execution_config=False, ) else: raise ValueError(f"Unknown agent type: {config.type}") self._agents[config.name] = agent return agent