custom agent poc

examples/custom-agents/agent.py

@@ -0,0 +1,22 @@
+import asyncio
+from mcp_agent.core.fastagent import FastAgent
+from mcp_agent.agents.base_agent import BaseAgent
+
+# Create the application
+fast = FastAgent("fast-agent example")
+
+class MyAgent(BaseAgent):
+    async def initialize(self):
+        await super().initialize()
+        print("it's a-me!...Mario!")
+
+# Define the agent
+@fast.custom(MyAgent, instruction="You are a helpful AI Agent")
+async def main():
+    # use the --model command line switch or agent arguments to change model
+    async with fast.run() as agent:
+        await agent.interactive()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/custom-agents/fastagent.config.yaml

@@ -0,0 +1,24 @@
+# Please edit this configuration file to match your environment (on Windows).
+# Examples in comments below - check/change the paths.
+#
+#
+
+execution_engine: asyncio
+logger:
+  type: file
+  level: error
+  truncate_tools: true
+
+mcp:
+  servers:
+    filesystem:
+      # On windows update the command and arguments to use `node` and the absolute path to the server.
+      # Use `npm i -g @modelcontextprotocol/server-filesystem` to install the server globally.
+      # Use `npm -g root` to find the global node_modules path.`
+      # command: "node"
+      # args: ["c:/Program Files/nodejs/node_modules/@modelcontextprotocol/server-filesystem/dist/index.js","."]
+      command: "npx"
+      args: ["-y", "@modelcontextprotocol/server-filesystem", "."]
+    fetch:
+      command: "uvx"
+      args: ["mcp-server-fetch"]

src/mcp_agent/core/agent_types.py

@@ -6,6 +6,7 @@
 from typing import List
 
 from pydantic import BaseModel, Field, model_validator
+from pydantic_core.core_schema import CustomErrorSchema
 
 # Forward imports to avoid circular dependencies
 from mcp_agent.core.request_params import RequestParams
@@ -15,6 +16,7 @@ class AgentType(Enum):
     """Enumeration of supported agent types."""
 
     BASIC = "agent"
+    CUSTOM = "custom"
     ORCHESTRATOR = "orchestrator"
     PARALLEL = "parallel"
     EVALUATOR_OPTIMIZER = "evaluator_optimizer"

src/mcp_agent/core/direct_decorators.py

@@ -205,10 +205,53 @@ def agent(
         human_input=human_input,
     )
 
+def custom(
+    self,
+    cls,
+    name: str = "default",
+    instruction_or_kwarg: Optional[str] = None,
+    *,
+    instruction: str = "You are a helpful agent.",
+    servers: List[str] = [],
+    model: Optional[str] = None,
+    use_history: bool = True,
+    request_params: RequestParams | None = None,
+    human_input: bool = False,
+) -> Callable[[AgentCallable[P, R]], DecoratedAgentProtocol[P, R]]:
+    """
+    Decorator to create and register a standard agent with type-safe signature.
+
+    Args:
+        name: Name of the agent
+        instruction_or_kwarg: Optional positional parameter for instruction
+        instruction: Base instruction for the agent (keyword arg)
+        servers: List of server names the agent should connect to
+        model: Model specification string
+        use_history: Whether to maintain conversation history
+        request_params: Additional request parameters for the LLM
+        human_input: Whether to enable human input capabilities
+
+    Returns:
+        A decorator that registers the agent with proper type annotations
+    """
+    final_instruction = instruction_or_kwarg if instruction_or_kwarg is not None else instruction
+
+    return _decorator_impl(
+        self,
+        AgentType.CUSTOM,
+        name=name,
+        instruction=final_instruction,
+        servers=servers,
+        model=model,
+        use_history=use_history,
+        request_params=request_params,
+        human_input=human_input,
+        agent_class=cls,
+    )
 
 DEFAULT_INSTRUCTION_ORCHESTRATOR = """
-You are an expert planner. Given an objective task and a list of Agents 
-(which are collections of capabilities), your job is to break down the objective 
+You are an expert planner. Given an objective task and a list of Agents
+(which are collections of capabilities), your job is to break down the objective
 into a series of steps, which can be performed by these agents.
 """
 
@@ -381,7 +424,7 @@ def parallel(
         A decorator that registers the parallel agent with proper type annotations
     """
     default_instruction = """
-    You are a parallel processor that executes multiple agents simultaneously 
+    You are a parallel processor that executes multiple agents simultaneously
     and aggregates their results.
     """
 

src/mcp_agent/core/direct_factory.py

@@ -153,6 +153,21 @@ def model_factory_func(model=None, request_params=None):
                 await agent.attach_llm(llm_factory, request_params=config.default_request_params)
                 result_agents[name] = agent
 
+            elif agent_type == AgentType.CUSTOM:
+                # Get the class to instantiate
+                cls = agent_data["agent_class"]
+                # Create the custom agent
+                agent = cls(
+                    config=config,
+                    context=app_instance.context,
+                )
+                await agent.initialize()
+
+                # Attach LLM to the agent
+                llm_factory = model_factory_func(model=config.model)
+                await agent.attach_llm(llm_factory, request_params=config.default_request_params)
+                result_agents[name] = agent
+
             elif agent_type == AgentType.ORCHESTRATOR:
                 # Get base params configured with model settings
                 base_params = (
@@ -355,6 +370,21 @@ async def create_agents_in_dependency_order(
             )
             active_agents.update(basic_agents)
 
+        # Create custom agents first
+        if AgentType.CUSTOM.value in [agents_dict[name]["type"] for name in group]:
+            basic_agents = await create_agents_by_type(
+                app_instance,
+                {
+                    name: agents_dict[name]
+                    for name in group
+                    if agents_dict[name]["type"] == AgentType.CUSTOM.value
+                },
+                AgentType.CUSTOM,
+                active_agents,
+                model_factory_func,
+            )
+            active_agents.update(basic_agents)
+
         # Create parallel agents
         if AgentType.PARALLEL.value in [agents_dict[name]["type"] for name in group]:
             parallel_agents = await create_agents_by_type(

src/mcp_agent/core/fastagent.py

@@ -20,6 +20,9 @@
 from mcp_agent.core.direct_decorators import (
     agent as agent_decorator,
 )
+from mcp_agent.core.direct_decorators import (
+    custom as custom_decorator,
+)
 from mcp_agent.core.direct_decorators import (
     chain as chain_decorator,
 )
@@ -197,6 +200,7 @@ def context(self) -> Context:
 
     # Decorator methods with type-safe implementations
     agent = agent_decorator
+    custom = custom_decorator
     orchestrator = orchestrator_decorator
     router = router_decorator
     chain = chain_decorator