Add Agent.to_mcp() method

Author: Kludex

pydantic_ai_slim/pydantic_ai/_mcp.py

@@ -1,11 +1,19 @@
 import base64
 from collections.abc import Sequence
-from typing import Literal
+from typing import Any, Literal, cast
+
+import logfire
+from pydantic.alias_generators import to_snake
+
+from pydantic_ai.agent.abstract import AbstractAgent
 
 from . import exceptions, messages
+from .agent import AgentDepsT, OutputDataT
 
 try:
     from mcp import types as mcp_types
+    from mcp.server.lowlevel.server import Server, StructuredContent
+    from mcp.types import Tool
 except ImportError as _import_error:
     raise ImportError(
         'Please install the `mcp` package to use the MCP server, '
@@ -121,3 +129,45 @@ def map_from_sampling_content(
         return messages.TextPart(content=content.text)
     else:
         raise NotImplementedError('Image and Audio responses in sampling are not yet supported')
+
+
+def agent_to_mcp(
+    agent: AbstractAgent[AgentDepsT, OutputDataT],
+    *,
+    server_name: str | None = None,
+    tool_name: str | None = None,
+    tool_description: str | None = None,
+    # TODO(Marcelo): Should this actually be a factory that is created in every tool call?
+    deps: AgentDepsT = None,
+) -> Server:
+    server_name = to_snake((server_name or agent.name or 'PydanticAI Agent').replace(' ', '_'))
+    tool_name = to_snake((tool_name or agent.name or 'PydanticAI Agent').replace(' ', '_'))
+    app = Server(name=server_name)
+
+    async def list_tools() -> list[Tool]:
+        return [
+            Tool(
+                name=tool_name,
+                description=tool_description,
+                inputSchema={'type': 'object', 'properties': {'prompt': {'type': 'string'}}},
+                # TODO(Marcelo): How do I get this?
+                outputSchema={'type': 'object', 'properties': {}},
+            )
+        ]
+
+    async def call_tool(name: str, args: dict[str, Any]) -> StructuredContent:
+        if name != tool_name:
+            raise ValueError(f'Unknown tool: {name}')
+
+        # TODO(Marcelo): Should we pass the `message_history` instead?
+        prompt = cast(str, args['prompt'])
+        logfire.info(f'Calling tool: {name} with args: {args}')
+
+        result = await agent.run(user_prompt=prompt, deps=deps)
+
+        return dict(result=result.output)
+
+    app.list_tools()(list_tools)
+    app.call_tool()(call_tool)
+
+    return app

pydantic_ai_slim/pydantic_ai/agent/abstract.py

@@ -1,27 +1,19 @@
 from __future__ import annotations as _annotations
 
 import inspect
+import warnings
 from abc import ABC, abstractmethod
 from collections.abc import AsyncIterable, AsyncIterator, Awaitable, Callable, Iterator, Mapping, Sequence
 from contextlib import AbstractAsyncContextManager, asynccontextmanager, contextmanager
 from types import FrameType
 from typing import TYPE_CHECKING, Any, Generic, TypeAlias, cast, overload
 
-from typing_extensions import Self, TypeIs, TypeVar
+from typing_extensions import TypeIs, TypeVar
 
 from pydantic_graph import End
 from pydantic_graph._utils import get_event_loop
 
-from .. import (
-    _agent_graph,
-    _system_prompt,
-    _utils,
-    exceptions,
-    messages as _messages,
-    models,
-    result,
-    usage as _usage,
-)
+from .. import _agent_graph, _system_prompt, _utils, exceptions, messages as _messages, models, result, usage as _usage
 from .._tool_manager import ToolManager
 from ..output import OutputDataT, OutputSpec
 from ..result import AgentStream, FinalResult, StreamedRunResult
@@ -42,6 +34,7 @@
     from fasta2a.broker import Broker
     from fasta2a.schema import AgentProvider, Skill
     from fasta2a.storage import Storage
+    from mcp.server.lowlevel import Server
     from starlette.middleware import Middleware
     from starlette.routing import BaseRoute, Route
     from starlette.types import ExceptionHandler, Lifespan
@@ -940,8 +933,28 @@ def to_a2a(
             lifespan=lifespan,
         )
 
+    def to_mcp(
+        self,
+        *,
+        server_name: str | None = None,
+        tool_name: str | None = None,
+        tool_description: str | None = None,
+        deps: AgentDepsT = None,
+    ) -> Server:
+        from .._mcp import agent_to_mcp
+
+        warnings.warn('The `to_mcp` method is experimental, and may change in the future.', UserWarning)
+
+        return agent_to_mcp(
+            self,
+            server_name=server_name,
+            tool_name=tool_name,
+            tool_description=tool_description,
+            deps=deps,
+        )
+
     async def to_cli(
-        self: Self,
+        self,
         deps: AgentDepsT = None,
         prog_name: str = 'pydantic-ai',
         message_history: list[_messages.ModelMessage] | None = None,
@@ -978,7 +991,7 @@ async def main():
         )
 
     def to_cli_sync(
-        self: Self,
+        self,
         deps: AgentDepsT = None,
         prog_name: str = 'pydantic-ai',
         message_history: list[_messages.ModelMessage] | None = None,