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