pydantic_ai.mcp

MCPServer

基类：AbstractToolset[Any], ABC

用于将代理附加到 MCP 服务器的基类。

更多信息请参见 https://modelcontextprotocol.net.cn。

源代码位于 pydantic_ai_slim/pydantic_ai/mcp.py

class MCPServer(AbstractToolset[Any], ABC):
    """Base class for attaching agents to MCP servers.

    See <https://modelcontextprotocol.net.cn> for more information.
    """

    tool_prefix: str | None
    """A prefix to add to all tools that are registered with the server.

    If not empty, will include a trailing underscore(`_`).

    e.g. if `tool_prefix='foo'`, then a tool named `bar` will be registered as `foo_bar`
    """

    log_level: mcp_types.LoggingLevel | None
    """The log level to set when connecting to the server, if any.

    See <https://modelcontextprotocol.net.cn/specification/2025-03-26/server/utilities/logging#logging> for more details.

    If `None`, no log level will be set.
    """

    log_handler: LoggingFnT | None
    """A handler for logging messages from the server."""

    timeout: float
    """The timeout in seconds to wait for the client to initialize."""

    read_timeout: float
    """Maximum time in seconds to wait for new messages before timing out.

    This timeout applies to the long-lived connection after it's established.
    If no new messages are received within this time, the connection will be considered stale
    and may be closed. Defaults to 5 minutes (300 seconds).
    """

    process_tool_call: ProcessToolCallback | None
    """Hook to customize tool calling and optionally pass extra metadata."""

    allow_sampling: bool
    """Whether to allow MCP sampling through this client."""

    sampling_model: models.Model | None
    """The model to use for sampling."""

    max_retries: int
    """The maximum number of times to retry a tool call."""

    elicitation_callback: ElicitationFnT | None = None
    """Callback function to handle elicitation requests from the server."""

    _id: str | None

    _enter_lock: Lock = field(compare=False)
    _running_count: int
    _exit_stack: AsyncExitStack | None

    _client: ClientSession
    _read_stream: MemoryObjectReceiveStream[SessionMessage | Exception]
    _write_stream: MemoryObjectSendStream[SessionMessage]

    def __init__(
        self,
        tool_prefix: str | None = None,
        log_level: mcp_types.LoggingLevel | None = None,
        log_handler: LoggingFnT | None = None,
        timeout: float = 5,
        read_timeout: float = 5 * 60,
        process_tool_call: ProcessToolCallback | None = None,
        allow_sampling: bool = True,
        sampling_model: models.Model | None = None,
        max_retries: int = 1,
        elicitation_callback: ElicitationFnT | None = None,
        *,
        id: str | None = None,
    ):
        self.tool_prefix = tool_prefix
        self.log_level = log_level
        self.log_handler = log_handler
        self.timeout = timeout
        self.read_timeout = read_timeout
        self.process_tool_call = process_tool_call
        self.allow_sampling = allow_sampling
        self.sampling_model = sampling_model
        self.max_retries = max_retries
        self.elicitation_callback = elicitation_callback

        self._id = id or tool_prefix

        self.__post_init__()

    def __post_init__(self):
        self._enter_lock = Lock()
        self._running_count = 0
        self._exit_stack = None

    @abstractmethod
    @asynccontextmanager
    async def client_streams(
        self,
    ) -> AsyncIterator[
        tuple[
            MemoryObjectReceiveStream[SessionMessage | Exception],
            MemoryObjectSendStream[SessionMessage],
        ]
    ]:
        """Create the streams for the MCP server."""
        raise NotImplementedError('MCP Server subclasses must implement this method.')
        yield

    @property
    def id(self) -> str | None:
        return self._id

    @property
    def label(self) -> str:
        if self.id:
            return super().label  # pragma: no cover
        else:
            return repr(self)

    @property
    def tool_name_conflict_hint(self) -> str:
        return 'Set the `tool_prefix` attribute to avoid name conflicts.'

    async def list_tools(self) -> list[mcp_types.Tool]:
        """Retrieve tools that are currently active on the server.

        Note:
        - We don't cache tools as they might change.
        - We also don't subscribe to the server to avoid complexity.
        """
        async with self:  # Ensure server is running
            result = await self._client.list_tools()
        return result.tools

    async def direct_call_tool(
        self,
        name: str,
        args: dict[str, Any],
        metadata: dict[str, Any] | None = None,
    ) -> ToolResult:
        """Call a tool on the server.

        Args:
            name: The name of the tool to call.
            args: The arguments to pass to the tool.
            metadata: Request-level metadata (optional)

        Returns:
            The result of the tool call.

        Raises:
            ModelRetry: If the tool call fails.
        """
        async with self:  # Ensure server is running
            try:
                result = await self._client.send_request(
                    mcp_types.ClientRequest(
                        mcp_types.CallToolRequest(
                            method='tools/call',
                            params=mcp_types.CallToolRequestParams(
                                name=name,
                                arguments=args,
                                _meta=mcp_types.RequestParams.Meta(**metadata) if metadata else None,
                            ),
                        )
                    ),
                    mcp_types.CallToolResult,
                )
            except McpError as e:
                raise exceptions.ModelRetry(e.error.message)

        content = [await self._map_tool_result_part(part) for part in result.content]

        if result.isError:
            text = '\n'.join(str(part) for part in content)
            raise exceptions.ModelRetry(text)
        else:
            return content[0] if len(content) == 1 else content

    async def call_tool(
        self,
        name: str,
        tool_args: dict[str, Any],
        ctx: RunContext[Any],
        tool: ToolsetTool[Any],
    ) -> ToolResult:
        if self.tool_prefix:
            name = name.removeprefix(f'{self.tool_prefix}_')
            ctx = replace(ctx, tool_name=name)

        if self.process_tool_call is not None:
            return await self.process_tool_call(ctx, self.direct_call_tool, name, tool_args)
        else:
            return await self.direct_call_tool(name, tool_args)

    async def get_tools(self, ctx: RunContext[Any]) -> dict[str, ToolsetTool[Any]]:
        return {
            name: self.tool_for_tool_def(
                ToolDefinition(
                    name=name,
                    description=mcp_tool.description,
                    parameters_json_schema=mcp_tool.inputSchema,
                ),
            )
            for mcp_tool in await self.list_tools()
            if (name := f'{self.tool_prefix}_{mcp_tool.name}' if self.tool_prefix else mcp_tool.name)
        }

    def tool_for_tool_def(self, tool_def: ToolDefinition) -> ToolsetTool[Any]:
        return ToolsetTool(
            toolset=self,
            tool_def=tool_def,
            max_retries=self.max_retries,
            args_validator=TOOL_SCHEMA_VALIDATOR,
        )

    async def __aenter__(self) -> Self:
        """Enter the MCP server context.

        This will initialize the connection to the server.
        If this server is an [`MCPServerStdio`][pydantic_ai.mcp.MCPServerStdio], the server will first be started as a subprocess.

        This is a no-op if the MCP server has already been entered.
        """
        async with self._enter_lock:
            if self._running_count == 0:
                async with AsyncExitStack() as exit_stack:
                    self._read_stream, self._write_stream = await exit_stack.enter_async_context(self.client_streams())
                    client = ClientSession(
                        read_stream=self._read_stream,
                        write_stream=self._write_stream,
                        sampling_callback=self._sampling_callback if self.allow_sampling else None,
                        elicitation_callback=self.elicitation_callback,
                        logging_callback=self.log_handler,
                        read_timeout_seconds=timedelta(seconds=self.read_timeout),
                    )
                    self._client = await exit_stack.enter_async_context(client)

                    with anyio.fail_after(self.timeout):
                        await self._client.initialize()

                        if log_level := self.log_level:
                            await self._client.set_logging_level(log_level)

                    self._exit_stack = exit_stack.pop_all()
            self._running_count += 1
        return self

    async def __aexit__(self, *args: Any) -> bool | None:
        if self._running_count == 0:
            raise ValueError('MCPServer.__aexit__ called more times than __aenter__')
        async with self._enter_lock:
            self._running_count -= 1
            if self._running_count == 0 and self._exit_stack is not None:
                await self._exit_stack.aclose()
                self._exit_stack = None

    @property
    def is_running(self) -> bool:
        """Check if the MCP server is running."""
        return bool(self._running_count)

    async def _sampling_callback(
        self, context: RequestContext[ClientSession, Any], params: mcp_types.CreateMessageRequestParams
    ) -> mcp_types.CreateMessageResult | mcp_types.ErrorData:
        """MCP sampling callback."""
        if self.sampling_model is None:
            raise ValueError('Sampling model is not set')  # pragma: no cover

        pai_messages = _mcp.map_from_mcp_params(params)
        model_settings = models.ModelSettings()
        if max_tokens := params.maxTokens:  # pragma: no branch
            model_settings['max_tokens'] = max_tokens
        if temperature := params.temperature:  # pragma: no branch
            model_settings['temperature'] = temperature
        if stop_sequences := params.stopSequences:  # pragma: no branch
            model_settings['stop_sequences'] = stop_sequences

        model_response = await model_request(self.sampling_model, pai_messages, model_settings=model_settings)
        return mcp_types.CreateMessageResult(
            role='assistant',
            content=_mcp.map_from_model_response(model_response),
            model=self.sampling_model.model_name,
        )

    async def _map_tool_result_part(
        self, part: mcp_types.ContentBlock
    ) -> str | messages.BinaryContent | dict[str, Any] | list[Any]:
        # See https://github.com/jlowin/fastmcp/blob/main/docs/servers/tools.mdx#return-values

        if isinstance(part, mcp_types.TextContent):
            text = part.text
            if text.startswith(('[', '{')):
                try:
                    return pydantic_core.from_json(text)
                except ValueError:
                    pass
            return text
        elif isinstance(part, mcp_types.ImageContent):
            return messages.BinaryContent(data=base64.b64decode(part.data), media_type=part.mimeType)
        elif isinstance(part, mcp_types.AudioContent):
            # NOTE: The FastMCP server doesn't support audio content.
            # See <https://github.com/modelcontextprotocol/python-sdk/issues/952> for more details.
            return messages.BinaryContent(
                data=base64.b64decode(part.data), media_type=part.mimeType
            )  # pragma: no cover
        elif isinstance(part, mcp_types.EmbeddedResource):
            resource = part.resource
            return self._get_content(resource)
        elif isinstance(part, mcp_types.ResourceLink):
            resource_result: mcp_types.ReadResourceResult = await self._client.read_resource(part.uri)
            return (
                self._get_content(resource_result.contents[0])
                if len(resource_result.contents) == 1
                else [self._get_content(resource) for resource in resource_result.contents]
            )
        else:
            assert_never(part)

    def _get_content(
        self, resource: mcp_types.TextResourceContents | mcp_types.BlobResourceContents
    ) -> str | messages.BinaryContent:
        if isinstance(resource, mcp_types.TextResourceContents):
            return resource.text
        elif isinstance(resource, mcp_types.BlobResourceContents):
            return messages.BinaryContent(
                data=base64.b64decode(resource.blob), media_type=resource.mimeType or 'application/octet-stream'
            )
        else:
            assert_never(resource)

tool_prefix instance-attribute

tool_prefix: str | None = tool_prefix

添加到所有在服务器上注册的工具的前缀。

如果不为空，将包含一个尾随下划线（_）。

例如，如果 tool_prefix='foo'，那么名为 bar 的工具将被注册为 foo_bar。

log_level instance-attribute

log_level: LoggingLevel | None = log_level

连接到服务器时设置的日志级别（如果有的话）。

更多详情请参见 https://modelcontextprotocol.net.cn/specification/2025-03-26/server/utilities/logging#logging。

如果为 None，则不会设置日志级别。

log_handler instance-attribute

log_handler: LoggingFnT | None = log_handler

用于处理来自服务器的日志消息的处理器。

timeout instance-attribute

timeout: float = timeout

等待客户端初始化的超时时间（秒）。

read_timeout instance-attribute

read_timeout: float = read_timeout

等待新消息的超时前的最大时间（秒）。

此超时适用于建立后的长连接。如果在此时间内没有收到新消息，连接将被视为空闲并可能被关闭。默认为 5 分钟（300 秒）。

process_tool_call instance-attribute

process_tool_call: ProcessToolCallback | None = (
    process_tool_call
)

用于自定义工具调用并可选地传递额外元数据的钩子。

allow_sampling instance-attribute

allow_sampling: bool = allow_sampling

是否允许通过此客户端进行 MCP 采样。

sampling_model instance-attribute

sampling_model: Model | None = sampling_model

用于采样的模型。

max_retries instance-attribute

max_retries: int = max_retries

重试工具调用的最大次数。

elicitation_callback class-attribute instance-attribute

elicitation_callback: ElicitationFnT | None = (
    elicitation_callback
)

用于处理来自服务器的启发请求的回调函数。

client_streams abstractmethod async

client_streams() -> AsyncIterator[
    tuple[
        MemoryObjectReceiveStream[
            SessionMessage | Exception
        ],
        MemoryObjectSendStream[SessionMessage],
    ]
]

为 MCP 服务器创建流。

源代码位于 pydantic_ai_slim/pydantic_ai/mcp.py

@abstractmethod
@asynccontextmanager
async def client_streams(
    self,
) -> AsyncIterator[
    tuple[
        MemoryObjectReceiveStream[SessionMessage | Exception],
        MemoryObjectSendStream[SessionMessage],
    ]
]:
    """Create the streams for the MCP server."""
    raise NotImplementedError('MCP Server subclasses must implement this method.')
    yield

list_tools async

list_tools() -> list[Tool]

检索服务器上当前活动的工具。

注意： - 我们不缓存工具，因为它们可能会改变。 - 我们也不订阅服务器以避免复杂性。

源代码位于 pydantic_ai_slim/pydantic_ai/mcp.py

async def list_tools(self) -> list[mcp_types.Tool]:
    """Retrieve tools that are currently active on the server.

    Note:
    - We don't cache tools as they might change.
    - We also don't subscribe to the server to avoid complexity.
    """
    async with self:  # Ensure server is running
        result = await self._client.list_tools()
    return result.tools

direct_call_tool async

direct_call_tool(
    name: str,
    args: dict[str, Any],
    metadata: dict[str, Any] | None = None,
) -> ToolResult

在服务器上调用一个工具。

参数

名称	类型	描述	默认值
name	str	要调用的工具的名称。	必需
args	dict[str, Any]	传递给工具的参数。	必需
metadata	dict[str, Any] | None	请求级别的元数据（可选）	None

返回

类型	描述
ToolResult	工具调用的结果。

引发

类型	描述
ModelRetry	如果工具调用失败。

源代码位于 pydantic_ai_slim/pydantic_ai/mcp.py

async def direct_call_tool(
    self,
    name: str,
    args: dict[str, Any],
    metadata: dict[str, Any] | None = None,
) -> ToolResult:
    """Call a tool on the server.

    Args:
        name: The name of the tool to call.
        args: The arguments to pass to the tool.
        metadata: Request-level metadata (optional)

    Returns:
        The result of the tool call.

    Raises:
        ModelRetry: If the tool call fails.
    """
    async with self:  # Ensure server is running
        try:
            result = await self._client.send_request(
                mcp_types.ClientRequest(
                    mcp_types.CallToolRequest(
                        method='tools/call',
                        params=mcp_types.CallToolRequestParams(
                            name=name,
                            arguments=args,
                            _meta=mcp_types.RequestParams.Meta(**metadata) if metadata else None,
                        ),
                    )
                ),
                mcp_types.CallToolResult,
            )
        except McpError as e:
            raise exceptions.ModelRetry(e.error.message)

    content = [await self._map_tool_result_part(part) for part in result.content]

    if result.isError:
        text = '\n'.join(str(part) for part in content)
        raise exceptions.ModelRetry(text)
    else:
        return content[0] if len(content) == 1 else content

__aenter__ 异步

__aenter__() -> Self

进入 MCP 服务器上下文。

这将初始化到服务器的连接。如果此服务器是 MCPServerStdio，服务器将首先作为子进程启动。

如果 MCP 服务器已经进入上下文，此操作为空操作。

源代码位于 pydantic_ai_slim/pydantic_ai/mcp.py

async def __aenter__(self) -> Self:
    """Enter the MCP server context.

    This will initialize the connection to the server.
    If this server is an [`MCPServerStdio`][pydantic_ai.mcp.MCPServerStdio], the server will first be started as a subprocess.

    This is a no-op if the MCP server has already been entered.
    """
    async with self._enter_lock:
        if self._running_count == 0:
            async with AsyncExitStack() as exit_stack:
                self._read_stream, self._write_stream = await exit_stack.enter_async_context(self.client_streams())
                client = ClientSession(
                    read_stream=self._read_stream,
                    write_stream=self._write_stream,
                    sampling_callback=self._sampling_callback if self.allow_sampling else None,
                    elicitation_callback=self.elicitation_callback,
                    logging_callback=self.log_handler,
                    read_timeout_seconds=timedelta(seconds=self.read_timeout),
                )
                self._client = await exit_stack.enter_async_context(client)

                with anyio.fail_after(self.timeout):
                    await self._client.initialize()

                    if log_level := self.log_level:
                        await self._client.set_logging_level(log_level)

                self._exit_stack = exit_stack.pop_all()
        self._running_count += 1
    return self

is_running property

is_running: bool

检查 MCP 服务器是否正在运行。

MCPServerStdio

基类：MCPServer

在一个子进程中运行一个 MCP 服务器，并通过标准输入/输出（stdin/stdout）与其通信。

此类实现了 MCP 规范中的 stdio 传输。更多信息请参见 https://spec.modelcontextprotocol.io/specification/2024-11-05/basic/transports/#stdio。

注意

将此类用作异步上下文管理器时，将在进入上下文时作为子进程启动服务器，并在退出上下文时停止它。

示例

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStdio

server = MCPServerStdio(  # (1)!
    'uv', args=['run', 'mcp-run-python', 'stdio'], timeout=10
)
agent = Agent('openai:gpt-4o', toolsets=[server])

async def main():
    async with agent:  # (2)!
        ...

1. 更多信息请参见 MCP Run Python。
2. 这将作为子进程启动服务器并连接到它。

源代码位于 pydantic_ai_slim/pydantic_ai/mcp.py

class MCPServerStdio(MCPServer):
    """Runs an MCP server in a subprocess and communicates with it over stdin/stdout.

    This class implements the stdio transport from the MCP specification.
    See <https://spec.modelcontextprotocol.io/specification/2024-11-05/basic/transports/#stdio> for more information.

    !!! note
        Using this class as an async context manager will start the server as a subprocess when entering the context,
        and stop it when exiting the context.

    Example:
    ```python {py="3.10"}
    from pydantic_ai import Agent
    from pydantic_ai.mcp import MCPServerStdio

    server = MCPServerStdio(  # (1)!
        'uv', args=['run', 'mcp-run-python', 'stdio'], timeout=10
    )
    agent = Agent('openai:gpt-4o', toolsets=[server])

    async def main():
        async with agent:  # (2)!
            ...
    ```

    1. See [MCP Run Python](https://github.com/pydantic/mcp-run-python) for more information.
    2. This will start the server as a subprocess and connect to it.
    """

    command: str
    """The command to run."""

    args: Sequence[str]
    """The arguments to pass to the command."""

    env: dict[str, str] | None
    """The environment variables the CLI server will have access to.

    By default the subprocess will not inherit any environment variables from the parent process.
    If you want to inherit the environment variables from the parent process, use `env=os.environ`.
    """

    cwd: str | Path | None
    """The working directory to use when spawning the process."""

    # last fields are re-defined from the parent class so they appear as fields
    tool_prefix: str | None
    log_level: mcp_types.LoggingLevel | None
    log_handler: LoggingFnT | None
    timeout: float
    read_timeout: float
    process_tool_call: ProcessToolCallback | None
    allow_sampling: bool
    sampling_model: models.Model | None
    max_retries: int
    elicitation_callback: ElicitationFnT | None = None

    def __init__(
        self,
        command: str,
        args: Sequence[str],
        *,
        env: dict[str, str] | None = None,
        cwd: str | Path | None = None,
        tool_prefix: str | None = None,
        log_level: mcp_types.LoggingLevel | None = None,
        log_handler: LoggingFnT | None = None,
        timeout: float = 5,
        read_timeout: float = 5 * 60,
        process_tool_call: ProcessToolCallback | None = None,
        allow_sampling: bool = True,
        sampling_model: models.Model | None = None,
        max_retries: int = 1,
        elicitation_callback: ElicitationFnT | None = None,
        id: str | None = None,
    ):
        """Build a new MCP server.

        Args:
            command: The command to run.
            args: The arguments to pass to the command.
            env: The environment variables to set in the subprocess.
            cwd: The working directory to use when spawning the process.
            tool_prefix: A prefix to add to all tools that are registered with the server.
            log_level: The log level to set when connecting to the server, if any.
            log_handler: A handler for logging messages from the server.
            timeout: The timeout in seconds to wait for the client to initialize.
            read_timeout: Maximum time in seconds to wait for new messages before timing out.
            process_tool_call: Hook to customize tool calling and optionally pass extra metadata.
            allow_sampling: Whether to allow MCP sampling through this client.
            sampling_model: The model to use for sampling.
            max_retries: The maximum number of times to retry a tool call.
            elicitation_callback: Callback function to handle elicitation requests from the server.
            id: An optional unique ID for the MCP server. An MCP server needs to have an ID in order to be used in a durable execution environment like Temporal, in which case the ID will be used to identify the server's activities within the workflow.
        """
        self.command = command
        self.args = args
        self.env = env
        self.cwd = cwd

        super().__init__(
            tool_prefix,
            log_level,
            log_handler,
            timeout,
            read_timeout,
            process_tool_call,
            allow_sampling,
            sampling_model,
            max_retries,
            elicitation_callback,
            id=id,
        )

    @asynccontextmanager
    async def client_streams(
        self,
    ) -> AsyncIterator[
        tuple[
            MemoryObjectReceiveStream[SessionMessage | Exception],
            MemoryObjectSendStream[SessionMessage],
        ]
    ]:
        server = StdioServerParameters(command=self.command, args=list(self.args), env=self.env, cwd=self.cwd)
        async with stdio_client(server=server) as (read_stream, write_stream):
            yield read_stream, write_stream

    def __repr__(self) -> str:
        repr_args = [
            f'command={self.command!r}',
            f'args={self.args!r}',
        ]
        if self.id:
            repr_args.append(f'id={self.id!r}')  # pragma: no cover
        return f'{self.__class__.__name__}({", ".join(repr_args)})'

__init__

__init__(
    command: str,
    args: Sequence[str],
    *,
    env: dict[str, str] | None = None,
    cwd: str | Path | None = None,
    tool_prefix: str | None = None,
    log_level: LoggingLevel | None = None,
    log_handler: LoggingFnT | None = None,
    timeout: float = 5,
    read_timeout: float = 5 * 60,
    process_tool_call: ProcessToolCallback | None = None,
    allow_sampling: bool = True,
    sampling_model: Model | None = None,
    max_retries: int = 1,
    elicitation_callback: ElicitationFnT | None = None,
    id: str | None = None
)

构建一个新的 MCP 服务器。

参数

名称	类型	描述	默认值
command	str	要运行的命令。	必需
args	Sequence[str]	传递给命令的参数。	必需
env	dict[str, str] | None	在子进程中设置的环境变量。	None
cwd	str | Path | None	启动进程时使用的工作目录。	None
tool_prefix	str | None	添加到所有在服务器上注册的工具的前缀。	None
log_level	LoggingLevel | None	连接到服务器时设置的日志级别（如果有的话）。	None
log_handler	LoggingFnT | None	用于处理来自服务器的日志消息的处理器。	None
timeout	float	等待客户端初始化的超时时间（秒）。	5
read_timeout	float	等待新消息的超时前的最大时间（秒）。	5 * 60
process_tool_call	ProcessToolCallback | None	用于自定义工具调用并可选地传递额外元数据的钩子。	None
allow_sampling	bool	是否允许通过此客户端进行 MCP 采样。	True
sampling_model	Model | None	用于采样的模型。	None
max_retries	int	重试工具调用的最大次数。	1
elicitation_callback	ElicitationFnT | None	用于处理来自服务器的启发请求的回调函数。	None
id	str | None	MCP 服务器的可选唯一 ID。MCP 服务器需要有一个 ID 才能在持久执行环境（如 Temporal）中使用，在这种情况下，ID 将用于标识工作流中服务器的活动。	None

源代码位于 pydantic_ai_slim/pydantic_ai/mcp.py

def __init__(
    self,
    command: str,
    args: Sequence[str],
    *,
    env: dict[str, str] | None = None,
    cwd: str | Path | None = None,
    tool_prefix: str | None = None,
    log_level: mcp_types.LoggingLevel | None = None,
    log_handler: LoggingFnT | None = None,
    timeout: float = 5,
    read_timeout: float = 5 * 60,
    process_tool_call: ProcessToolCallback | None = None,
    allow_sampling: bool = True,
    sampling_model: models.Model | None = None,
    max_retries: int = 1,
    elicitation_callback: ElicitationFnT | None = None,
    id: str | None = None,
):
    """Build a new MCP server.

    Args:
        command: The command to run.
        args: The arguments to pass to the command.
        env: The environment variables to set in the subprocess.
        cwd: The working directory to use when spawning the process.
        tool_prefix: A prefix to add to all tools that are registered with the server.
        log_level: The log level to set when connecting to the server, if any.
        log_handler: A handler for logging messages from the server.
        timeout: The timeout in seconds to wait for the client to initialize.
        read_timeout: Maximum time in seconds to wait for new messages before timing out.
        process_tool_call: Hook to customize tool calling and optionally pass extra metadata.
        allow_sampling: Whether to allow MCP sampling through this client.
        sampling_model: The model to use for sampling.
        max_retries: The maximum number of times to retry a tool call.
        elicitation_callback: Callback function to handle elicitation requests from the server.
        id: An optional unique ID for the MCP server. An MCP server needs to have an ID in order to be used in a durable execution environment like Temporal, in which case the ID will be used to identify the server's activities within the workflow.
    """
    self.command = command
    self.args = args
    self.env = env
    self.cwd = cwd

    super().__init__(
        tool_prefix,
        log_level,
        log_handler,
        timeout,
        read_timeout,
        process_tool_call,
        allow_sampling,
        sampling_model,
        max_retries,
        elicitation_callback,
        id=id,
    )

command instance-attribute

command: str = command

要运行的命令。

args instance-attribute

args: Sequence[str] = args

传递给命令的参数。

env instance-attribute

env: dict[str, str] | None = env

CLI 服务器将有权访问的环境变量。

默认情况下，子进程不会从父进程继承任何环境变量。如果要从父进程继承环境变量，请使用 env=os.environ。

cwd instance-attribute

cwd: str | Path | None = cwd

启动进程时使用的工作目录。

MCPServerSSE

基类：_MCPServerHTTP

一个通过可流式 HTTP 连接进行连接的 MCP 服务器。

此类实现了 MCP 规范中的 SSE 传输。更多信息请参见 https://spec.modelcontextprotocol.io/specification/2024-11-05/basic/transports/#http-with-sse。

注意

将此类用作异步上下文管理器将创建一个新的 HTTP 连接池，以连接到一个应该已经在运行的服务器。

示例

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerSSE

server = MCPServerSSE('https://:3001/sse')
agent = Agent('openai:gpt-4o', toolsets=[server])

async def main():
    async with agent:  # (1)!
        ...

1. 这将连接到运行在 localhost:3001 上的服务器。

源代码位于 pydantic_ai_slim/pydantic_ai/mcp.py

class MCPServerSSE(_MCPServerHTTP):
    """An MCP server that connects over streamable HTTP connections.

    This class implements the SSE transport from the MCP specification.
    See <https://spec.modelcontextprotocol.io/specification/2024-11-05/basic/transports/#http-with-sse> for more information.

    !!! note
        Using this class as an async context manager will create a new pool of HTTP connections to connect
        to a server which should already be running.

    Example:
    ```python {py="3.10"}
    from pydantic_ai import Agent
    from pydantic_ai.mcp import MCPServerSSE

    server = MCPServerSSE('https://:3001/sse')
    agent = Agent('openai:gpt-4o', toolsets=[server])

    async def main():
        async with agent:  # (1)!
            ...
    ```

    1. This will connect to a server running on `localhost:3001`.
    """

    @property
    def _transport_client(self):
        return sse_client  # pragma: no cover

MCPServerHTTP deprecated

基类：MCPServerSSE

已弃用

MCPServerHTTP 类已弃用，请改用 MCPServerSSE。

一个使用旧的 SSE 传输通过 HTTP 连接的 MCP 服务器。

此类实现了 MCP 规范中的 SSE 传输。更多信息请参见 https://spec.modelcontextprotocol.io/specification/2024-11-05/basic/transports/#http-with-sse。

注意

将此类用作异步上下文管理器将创建一个新的 HTTP 连接池，以连接到一个应该已经在运行的服务器。

示例

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerHTTP

server = MCPServerHTTP('https://:3001/sse')
agent = Agent('openai:gpt-4o', toolsets=[server])

async def main():
    async with agent:  # (2)!
        ...

1. 这将连接到运行在 localhost:3001 上的服务器。

源代码位于 pydantic_ai_slim/pydantic_ai/mcp.py

@deprecated('The `MCPServerHTTP` class is deprecated, use `MCPServerSSE` instead.')
class MCPServerHTTP(MCPServerSSE):
    """An MCP server that connects over HTTP using the old SSE transport.

    This class implements the SSE transport from the MCP specification.
    See <https://spec.modelcontextprotocol.io/specification/2024-11-05/basic/transports/#http-with-sse> for more information.

    !!! note
        Using this class as an async context manager will create a new pool of HTTP connections to connect
        to a server which should already be running.

    Example:
    ```python {py="3.10" test="skip"}
    from pydantic_ai import Agent
    from pydantic_ai.mcp import MCPServerHTTP

    server = MCPServerHTTP('https://:3001/sse')
    agent = Agent('openai:gpt-4o', toolsets=[server])

    async def main():
        async with agent:  # (2)!
            ...
    ```

    1. This will connect to a server running on `localhost:3001`.
    """

MCPServerStreamableHTTP

基类：_MCPServerHTTP

一个使用可流式 HTTP 传输通过 HTTP 连接的 MCP 服务器。

此类实现了 MCP 规范中的可流式 HTTP 传输。更多信息请参见 https://modelcontextprotocol.net.cn/introduction#streamable-http。

注意

将此类用作异步上下文管理器将创建一个新的 HTTP 连接池，以连接到一个应该已经在运行的服务器。

示例

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP

server = MCPServerStreamableHTTP('https://:8000/mcp')  # (1)!
agent = Agent('openai:gpt-4o', toolsets=[server])

async def main():
    async with agent:  # (2)!
        ...

源代码位于 pydantic_ai_slim/pydantic_ai/mcp.py

class MCPServerStreamableHTTP(_MCPServerHTTP):
    """An MCP server that connects over HTTP using the Streamable HTTP transport.

    This class implements the Streamable HTTP transport from the MCP specification.
    See <https://modelcontextprotocol.net.cn/introduction#streamable-http> for more information.

    !!! note
        Using this class as an async context manager will create a new pool of HTTP connections to connect
        to a server which should already be running.

    Example:
    ```python {py="3.10"}
    from pydantic_ai import Agent
    from pydantic_ai.mcp import MCPServerStreamableHTTP

    server = MCPServerStreamableHTTP('https://:8000/mcp')  # (1)!
    agent = Agent('openai:gpt-4o', toolsets=[server])

    async def main():
        async with agent:  # (2)!
            ...
    ```
    """

    @property
    def _transport_client(self):
        return streamablehttp_client  # pragma: no cover