jianmu.mcp

适用对象：MCP 集成开发者 / 平台维护者 是否必读：按需 相关模块：jianmu.tool, jianmu.swarm

1. 模块职责

jianmu.mcp 提供对 MCP server / client / provider 的公开封装，让外部工具源可以被 jianmu 运行时消费。

它解决的是“把 MCP 工具接入 jianmu”这个问题，而不是本地内置工具声明问题。

2. 适合查什么

• 服务端配置与客户端：MCPServerConfig、MCPClient
• MCP 工具封装：MCPTool、MCPResourceTool
• provider 与 server 构造：MCPToolProvider、MCPServerBuilder

3. 使用建议

• 需要消费外部 MCP 工具时，从这里开始
• 如果只是写普通 Python 工具，应优先看 jianmu.tool

4. 最小示例

from jianmu.mcp import MCPClient, MCPServerConfig

config = MCPServerConfig(
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem"],
)
client = MCPClient(config)

5. 常见入口

• 想连 MCP server：看 MCPClient
• 想把 MCP 工具接进 tool provider：看 MCPToolProvider
• 想搭建 MCP server：看 MCPServer、MCPServerBuilder

6. 注意事项

• 如果只是写普通 Python 工具，优先回到 jianmu.tool
• MCP 的实际可用性依赖外部 server 进程和对应配置
• 这层解决的是“接入外部工具生态”，不是替代本地工具抽象

7. API 参考

mcp

Model Context Protocol (MCP) clients, providers, and server integrations for Jianmu.

MCPServerConfig dataclass

MCPServerConfig(
    command: str,
    args: List[str] = list(),
    env: Optional[Dict[str, str]] = None,
)

Configuration for starting a subprocess stdio MCP server.

属性：

名称	类型	描述
command	str	Executable name or absolute path used to start the MCP server.
args	List[str]	Positional arguments passed to the server process.
env	Optional[Dict[str, str]]	Optional environment variables merged into the child process.

MCPClient

MCPClient(
    server_source: Union[
        MCPServerConfig,
        str,
        List[str],
        FastMCP,
        Dict[str, Any],
    ],
    server_args: Optional[List[str]] = None,
    transport_type: Optional[str] = None,
    env: Optional[Dict[str, str]] = None,
    **transport_kwargs: Any,
)

MCP client with multi-transport support (fastmcp v2).

属性：

名称	类型	描述
server_args		Extra positional arguments appended to subprocess launches.
transport_type		Optional transport override such as stdio or sse.
env		Environment variables merged into subprocess transports.
transport_kwargs		Extra transport-specific keyword arguments.
server_source		Normalized FastMCP server or transport source.
client	Optional[Client]	Connected FastMCP client instance, if any.
_context_manager		Async context manager used for connection lifecycle.

Normalize server config and prepare a lazy MCP client.

参数：

名称	类型	描述	默认
server_source	Union[MCPServerConfig, str, List[str], FastMCP, Dict[str, Any]]	Server definition to connect to. This may be a MCPServerConfig, URL string, command list, FastMCP instance, or raw transport config dictionary.	必需
server_args	Optional[List[str]]	Extra positional arguments appended to stdio/python server launches.	None
transport_type	Optional[str]	Optional transport override such as http or sse when the source can map to multiple transport kinds.	None
env	Optional[Dict[str, str]]	Optional environment variables merged into subprocess transports.	None
transport_kwargs	Any	Extra transport-specific keyword arguments forwarded to the underlying FastMCP transport object.	{}

源代码位于： jianmu/mcp/client.py

def __init__(
    self,
    server_source: Union[MCPServerConfig, str, List[str], "FastMCP", Dict[str, Any]],
    server_args: Optional[List[str]] = None,
    transport_type: Optional[str] = None,
    env: Optional[Dict[str, str]] = None,
    **transport_kwargs: Any,
):
    """Normalize server config and prepare a lazy MCP client.

    Args:
        server_source: Server definition to connect to. This may be a
            ``MCPServerConfig``, URL string, command list, ``FastMCP``
            instance, or raw transport config dictionary.
        server_args: Extra positional arguments appended to stdio/python
            server launches.
        transport_type: Optional transport override such as ``http`` or ``sse``
            when the source can map to multiple transport kinds.
        env: Optional environment variables merged into subprocess transports.
        transport_kwargs: Extra transport-specific keyword arguments forwarded
            to the underlying FastMCP transport object.
    """
    if not FASTMCP_AVAILABLE:
        raise RuntimeError(
            "fastmcp package not installed. Run: pip install fastmcp>=2.0.0"
        )

    self.server_args = server_args or []
    self.transport_type = transport_type
    self.env = env or {}
    self.transport_kwargs = transport_kwargs
    self.server_source = self._prepare_server_source(server_source)
    self.client: Optional[Client] = None
    self._context_manager = None

connect async

connect() -> Any

Open the MCP transport and cache the connected client.

返回：

类型	描述
Any	The connected Client instance.

引发：

类型	描述
Exception	If connection to the MCP server fails.

源代码位于： jianmu/mcp/client.py

async def connect(self) -> Any:
    """Open the MCP transport and cache the connected client.

    Returns:
        The connected Client instance.

    Raises:
        Exception: If connection to the MCP server fails.
    """
    if self.client is not None:
        return self.client

    client = Client(self.server_source)
    try:
        await client.__aenter__()
    except Exception as exc:
        try:
            await client.__aexit__(type(exc), exc, exc.__traceback__)
        except Exception:
            pass
        raise

    self.client = client
    self._context_manager = client
    logger.debug("🔌 [MCP] Connected to server.")
    return self.client

close async

close()

Close the MCP connection and cleanup subprocess transport.

This method handles the cleanup of subprocess transports to avoid 'Event loop is closed' errors during garbage collection.

源代码位于： jianmu/mcp/client.py

async def close(self):
    """Close the MCP connection and cleanup subprocess transport.

    This method handles the cleanup of subprocess transports to avoid
    'Event loop is closed' errors during garbage collection.
    """
    if self._context_manager is None:
        return

    # Capture transport reference before closing
    transport = None
    if self.client is not None:
        transport = getattr(self.client, "transport", None)

    try:
        await self._context_manager.__aexit__(None, None, None)
    except Exception:
        pass  # Ignore errors during exit
    finally:
        self.client = None
        self._context_manager = None

    # For stdio transports (subprocesses), ensure clean shutdown
    if transport is not None:
        try:
            # Some transports have their own close method
            close_fn = getattr(transport, "close", None)
            if callable(close_fn):
                import asyncio
                result = close_fn()
                if asyncio.iscoroutine(result):
                    await result

            # Give subprocess a moment to exit cleanly
            import asyncio
            await asyncio.sleep(0.1)
        except Exception:
            pass  # Ignore cleanup errors

list_tools async

list_tools() -> List[Any]

List MCP tools exposed by the connected server.

返回：

类型	描述
List[Any]	The resulting List[Any] value.

源代码位于： jianmu/mcp/client.py

async def list_tools(self) -> List[Any]:
    """List MCP tools exposed by the connected server.

    Returns:
        The resulting `List[Any]` value.
    """
    client = await self.connect()
    response = await client.list_tools()
    if hasattr(response, "tools"):
        return response.tools
    if isinstance(response, list):
        return response
    return []

list_resources async

list_resources() -> List[Any]

List MCP resources exposed by the connected server.

返回：

类型	描述
List[Any]	The resulting List[Any] value.

源代码位于： jianmu/mcp/client.py

async def list_resources(self) -> List[Any]:
    """List MCP resources exposed by the connected server.

    Returns:
        The resulting `List[Any]` value.
    """
    client = await self.connect()
    response = await client.list_resources()
    if hasattr(response, "resources"):
        return response.resources
    if isinstance(response, list):
        return response
    return []

read_resource async

read_resource(uri: str) -> Any

Read one MCP resource by URI.

参数：

名称	类型	描述	默认
uri	str	The resource URI to read.	必需

返回：

类型	描述
Any	The resulting Any value.

源代码位于： jianmu/mcp/client.py

async def read_resource(self, uri: str) -> Any:
    """Read one MCP resource by URI.

    Args:
        uri: The resource URI to read.

    Returns:
        The resulting `Any` value.
    """
    client = await self.connect()
    return await client.read_resource(uri)

call_tool async

call_tool(
    name: str, arguments: Optional[Dict[str, Any]] = None
) -> Any

Invoke one MCP tool with optional arguments.

参数：

名称	类型	描述	默认
name	str	The name of the tool to invoke.	必需
arguments	Optional[Dict[str, Any]]	Optional dictionary of arguments for the tool call. Defaults to None.	None

返回：

类型	描述
Any	The resulting Any value.

源代码位于： jianmu/mcp/client.py

async def call_tool(self, name: str, arguments: Optional[Dict[str, Any]] = None) -> Any:
    """Invoke one MCP tool with optional arguments.

    Args:
        name: The name of the tool to invoke.
        arguments: Optional dictionary of arguments for the tool call. Defaults to None.

    Returns:
        The resulting `Any` value.
    """
    client = await self.connect()
    call_tool_mcp = getattr(client, "call_tool_mcp", None)
    if callable(call_tool_mcp):
        return await call_tool_mcp(name, arguments or {})
    return await client.call_tool(name, arguments or {})

list_prompts async

list_prompts() -> List[Any]

List MCP prompts exposed by the connected server.

返回：

类型	描述
List[Any]	The resulting List[Any] value.

源代码位于： jianmu/mcp/client.py

async def list_prompts(self) -> List[Any]:
    """List MCP prompts exposed by the connected server.

    Returns:
        The resulting `List[Any]` value.
    """
    client = await self.connect()
    response = await client.list_prompts()
    if hasattr(response, "prompts"):
        return response.prompts
    if isinstance(response, list):
        return response
    return []

get_prompt async

get_prompt(
    name: str, arguments: Optional[Dict[str, str]] = None
) -> Any

Fetch one MCP prompt definition by name.

参数：

名称	类型	描述	默认
name	str	The name of the prompt.	必需
arguments	Optional[Dict[str, str]]	Optional dictionary of arguments. Defaults to None.	None

返回：

类型	描述
Any	The resulting Any value.

源代码位于： jianmu/mcp/client.py

async def get_prompt(self, name: str, arguments: Optional[Dict[str, str]] = None) -> Any:
    """Fetch one MCP prompt definition by name.

    Args:
        name: The name of the prompt.
        arguments: Optional dictionary of arguments. Defaults to None.

    Returns:
        The resulting `Any` value.
    """
    client = await self.connect()
    return await client.get_prompt(name, arguments or {})

ping async

ping() -> bool

Ping the MCP server and report connectivity.

返回：

类型	描述
bool	True if the ping succeeded, False otherwise.

源代码位于： jianmu/mcp/client.py

async def ping(self) -> bool:
    """Ping the MCP server and report connectivity.

    Returns:
        True if the ping succeeded, False otherwise.
    """
    client = await self.connect()
    try:
        await client.ping()
        return True
    except Exception:
        return False

get_transport_info

get_transport_info() -> Dict[str, Any]

Return human-readable transport diagnostics for the current client.

返回：

类型	描述
Dict[str, Any]	A dictionary containing transport status and diagnostics.

源代码位于： jianmu/mcp/client.py

def get_transport_info(self) -> Dict[str, Any]:
    """Return human-readable transport diagnostics for the current client.

    Returns:
        A dictionary containing transport status and diagnostics.
    """
    if not self.client:
        return {"status": "not_connected"}
    transport = getattr(self.client, "transport", None)
    if transport:
        return {
            "status": "connected",
            "transport_type": type(transport).__name__,
            "transport_info": str(transport),
        }
    return {"status": "unknown"}

as_tools async

as_tools(
    allowlist: Optional[List[str]] = None,
) -> List[MCPTool]

Wrap MCP tool definitions as Jianmu tools.

参数：

名称	类型	描述	默认
allowlist	Optional[List[str]]	Optional list of tool names to include. If None, all tools are wrapped.	None

返回：

类型	描述
List[MCPTool]	A list of wrapped MCPTool objects.

源代码位于： jianmu/mcp/client.py

async def as_tools(self, allowlist: Optional[List[str]] = None) -> List["MCPTool"]:
    """Wrap MCP tool definitions as Jianmu tools.

    Args:
        allowlist: Optional list of tool names to include. If None, all tools are wrapped.

    Returns:
        A list of wrapped MCPTool objects.
    """
    tools = await self.list_tools()
    allow = set(t.lower() for t in allowlist) if allowlist else None
    result = []
    for tool in tools:
        if allow and tool.name.lower() not in allow:
            continue
        result.append(MCPTool(self, tool))
    return result

as_resource_tools async

as_resource_tools(
    allowlist: Optional[List[str]] = None,
) -> List[MCPResourceTool]

Wrap MCP resources as read-only Jianmu tools.

参数：

名称	类型	描述	默认
allowlist	Optional[List[str]]	Optional list of resource URIs to include. If None, all resources are wrapped.	None

返回：

类型	描述
List[MCPResourceTool]	A list of wrapped MCPResourceTool objects.

源代码位于： jianmu/mcp/client.py

async def as_resource_tools(self, allowlist: Optional[List[str]] = None) -> List["MCPResourceTool"]:
    """Wrap MCP resources as read-only Jianmu tools.

    Args:
        allowlist: Optional list of resource URIs to include. If None, all resources are wrapped.

    Returns:
        A list of wrapped MCPResourceTool objects.
    """
    resources = await self.list_resources()
    allow = set(r.lower() for r in allowlist) if allowlist else None
    result = []
    for resource in resources:
        uri = getattr(resource, "uri", "")
        if allow and uri.lower() not in allow:
            continue
        result.append(MCPResourceTool(self, resource))
    return result

MCPTool

MCPTool(client: MCPClient, tool_def: Any)

Bases: Tool

Wrapper that exposes an MCP tool definition as a standard Jianmu Tool.

属性：

名称	类型	描述
_client		MCP client used to invoke the wrapped tool.
_tool_def		Raw MCP tool definition returned by the server.
name		Public tool name exposed to Jianmu callers.
description		Human-readable tool description.
input_schema		JSON schema describing accepted inputs.
output_schema		JSON schema describing tool outputs.

Wrap one MCP tool definition as a Jianmu tool.

参数：

名称	类型	描述	默认
client	MCPClient	Connected or lazily connecting MCP client used to invoke the tool.	必需
tool_def	Any	Raw MCP tool definition returned by the server.	必需

源代码位于： jianmu/mcp/client.py

def __init__(self, client: MCPClient, tool_def: Any):
    """Wrap one MCP tool definition as a Jianmu tool.

    Args:
        client: Connected or lazily connecting MCP client used to invoke the tool.
        tool_def: Raw MCP tool definition returned by the server.
    """
    self._client = client
    self._tool_def = tool_def
    self.name = getattr(tool_def, "name", "mcp_tool")
    desc = getattr(tool_def, "description", "") or ""
    self.description = desc or self.name
    # Some OpenAI-compatible APIs reject JSON Schema metadata fields such as ``$schema``.
    raw_input = getattr(tool_def, "inputSchema", {"type": "object"})
    self.input_schema = self._clean_schema(raw_input)
    raw_output = getattr(tool_def, "outputSchema", {"type": "string"})
    self.output_schema = self._clean_schema(raw_output)

get_description

get_description() -> str

Return the MCP-provided tool description instead of the base docstring.

返回：

类型	描述
str	The tool description string.

源代码位于： jianmu/mcp/client.py

def get_description(self) -> str:
    """Return the MCP-provided tool description instead of the base docstring.

    Returns:
        The tool description string.
    """
    return self.description

run async

run(input: Any = None, **kwargs: Any) -> Any

Invoke the wrapped MCP tool.

参数：

名称	类型	描述	默认
input	Any	Single argument payload, or dictionary of arguments. Defaults to None.	None
kwargs	Any	Named arguments for the tool call.	{}

返回：

类型	描述
Any	The tool execution result, or an error string.

源代码位于： jianmu/mcp/client.py

async def run(self, input: Any = None, **kwargs: Any) -> Any:
    """Invoke the wrapped MCP tool.

    Args:
        input: Single argument payload, or dictionary of arguments. Defaults to None.
        kwargs: Named arguments for the tool call.

    Returns:
        The tool execution result, or an error string.
    """
    # If keyword arguments are provided, use them directly
    if kwargs:
        args = kwargs
    elif input is None:
        args = {}
    elif isinstance(input, dict):
        args = input
    else:
        args = {"input": input}
        if isinstance(self.input_schema, dict):
            props = self.input_schema.get("properties")
            if isinstance(props, dict) and len(props) == 1:
                key = next(iter(props.keys()))
                args = {key: input}

    result = await self._client.call_tool(self.name, arguments=args)

    if getattr(result, "isError", False):
        content = getattr(result, "content", None)
        if isinstance(content, list):
            for block in content:
                text = getattr(block, "text", None)
                if text is not None:
                    return f"Error: {text}"
        return "Error: MCP tool execution failed"

    structured = getattr(result, "structuredContent", None)
    if structured is None:
        structured = getattr(result, "structured_content", None)
    if structured is not None:
        return structured

    content = getattr(result, "content", None)
    if isinstance(content, list):
        for block in content:
            text = getattr(block, "text", None)
            if text is not None:
                return text
            data = getattr(block, "data", None)
            if data is not None:
                return data
            blob = getattr(block, "blob", None)
            if blob is not None:
                return blob

    return str(result)

MCPResourceTool

MCPResourceTool(client: MCPClient, resource_def: Any)

Bases: Tool

Read-only Tool wrapper for MCP resources.

属性：

名称	类型	描述
_client		MCP client used to read the wrapped resource.
_resource_def		Raw MCP resource definition returned by the server.
name		Public tool name exposed to Jianmu callers.
description		Human-readable resource description.
input_schema		Schema for optional URI override input.
output_schema		Schema describing returned resource content.

Wrap one MCP resource definition as a read-only tool.

参数：

名称	类型	描述	默认
client	MCPClient	Connected or lazily connecting MCP client used to read the resource.	必需
resource_def	Any	Raw MCP resource definition returned by the server.	必需

源代码位于： jianmu/mcp/client.py

def __init__(self, client: MCPClient, resource_def: Any):
    """Wrap one MCP resource definition as a read-only tool.

    Args:
        client: Connected or lazily connecting MCP client used to read the resource.
        resource_def: Raw MCP resource definition returned by the server.
    """
    self._client = client
    self._resource_def = resource_def
    self.name = getattr(resource_def, "name", None) or getattr(resource_def, "uri", "mcp_resource")
    self.description = getattr(resource_def, "description", "") or ""
    self.input_schema = {"type": "string", "description": "Resource URI (optional override)"}
    self.output_schema = {"type": "string"}

run async

run(input: Any) -> Any

Read the wrapped MCP resource and return its content.

参数：

名称	类型	描述	默认
input	Any	Resource URI override (optional).	必需

返回：

类型	描述
Any	The resource content, or an error string.

源代码位于： jianmu/mcp/client.py

async def run(self, input: Any) -> Any:
    """Read the wrapped MCP resource and return its content.

    Args:
        input: Resource URI override (optional).

    Returns:
        The resource content, or an error string.
    """
    uri = getattr(self._resource_def, "uri", None)
    if isinstance(input, str) and input:
        uri = input
    if not uri:
        return "Error: resource URI not provided"
    result = await self._client.read_resource(uri)

    contents = getattr(result, "contents", None)
    if isinstance(contents, list):
        for item in contents:
            text = getattr(item, "text", None)
            if text is not None:
                return text
            data = getattr(item, "data", None)
            if data is not None:
                return data
            blob = getattr(item, "blob", None)
            if blob is not None:
                return blob
    return str(result)

MCPToolProvider

MCPToolProvider(config_path: str = 'mcp.json')

Bases: BaseToolProvider

Load MCP tools from config and expose them as Jianmu tools.

属性：

名称	类型	描述
config_path		Filesystem path to the MCP config file.
_clients	dict[str, MCPClient]	Connected MCP clients keyed by configured server name.
_tools	list[Tool]	Cached wrapped Tool instances exported by all connected servers.

Create a provider backed by an MCP config file.

源代码位于： jianmu/mcp/provider.py

def __init__(self, config_path: str = "mcp.json"):
    """Create a provider backed by an MCP config file."""
    self.config_path = config_path
    self._clients: dict[str, MCPClient] = {}
    self._tools: list[Tool] = []

initialize async

initialize() -> None

Connect configured MCP clients eagerly.

源代码位于： jianmu/mcp/provider.py

async def initialize(self) -> None:
    """Connect configured MCP clients eagerly."""
    config = self._load_config()
    if not config:
        return
    servers = config.get("mcpServers", {}) if isinstance(config, dict) else {}
    if not isinstance(servers, dict):
        return
    for server_name, server_config in servers.items():
        if not isinstance(server_config, dict):
            continue
        try:
            await self._connect_server(server_name, server_config)
        except Exception as exc:
            logger.warning("⚠️ Failed to connect MCP server {}: {}", server_name, exc)

get_tools

get_tools(**_: Any) -> list[Tool]

Return cached MCP-backed Jianmu tools.

参数：

名称	类型	描述	默认
_	Any	Unused keyword arguments.	{}

返回：

类型	描述
list[Tool]	A list of cached Tool instances.

源代码位于： jianmu/mcp/provider.py

def get_tools(self, **_: Any) -> list[Tool]:
    """Return cached MCP-backed Jianmu tools.

    Args:
        _: Unused keyword arguments.

    Returns:
        A list of cached Tool instances.
    """
    return list(self._tools)

close async

close() -> None

Close all managed MCP clients.

源代码位于： jianmu/mcp/provider.py

async def close(self) -> None:
    """Close all managed MCP clients."""
    for client in self._clients.values():
        try:
            await client.close()
        except Exception as exc:
            logger.warning("⚠️ Failed to close MCP client: {}", exc)
    self._clients.clear()
    self._tools.clear()

MCPServer

MCPServer(name: str, description: Optional[str] = None)

Thin wrapper around FastMCP for convenience.

属性：

名称	类型	描述
mcp		Underlying FastMCP server instance.
name		Public server name exposed over the MCP protocol.
description		Human-readable server description.

Create a thin FastMCP-backed server wrapper.

源代码位于： jianmu/mcp/server.py

def __init__(self, name: str, description: Optional[str] = None):
    """Create a thin FastMCP-backed server wrapper."""
    _require_fastmcp()
    self.mcp = FastMCP(name=name)
    self.name = name
    self.description = description or f"{name} MCP Server"

add_tool

add_tool(
    func: Callable,
    name: Optional[str] = None,
    description: Optional[str] = None,
)

Register one tool implementation with the server.

参数：

名称	类型	描述	默认
func	Callable	The Python function that implements the tool.	必需
name	Optional[str]	Optional explicit tool name override. Defaults to None.	None
description	Optional[str]	Optional explicit tool description override. Defaults to None.	None

源代码位于： jianmu/mcp/server.py

def add_tool(self, func: Callable, name: Optional[str] = None, description: Optional[str] = None):
    """Register one tool implementation with the server.

    Args:
        func: The Python function that implements the tool.
        name: Optional explicit tool name override. Defaults to None.
        description: Optional explicit tool description override. Defaults to None.
    """
    if name or description:
        self.mcp.tool(name=name, description=description)(func)
    else:
        self.mcp.tool()(func)

add_resource

add_resource(
    func: Callable,
    uri: Optional[str] = None,
    name: Optional[str] = None,
    description: Optional[str] = None,
)

Register one MCP resource handler.

参数：

名称	类型	描述	默认
func	Callable	The Python function that resolves the resource content.	必需
uri	Optional[str]	Optional resource URI template. Defaults to None.	None
name	Optional[str]	Optional resource name. Defaults to None.	None
description	Optional[str]	Optional resource description. Defaults to None.	None

源代码位于： jianmu/mcp/server.py

def add_resource(
    self,
    func: Callable,
    uri: Optional[str] = None,
    name: Optional[str] = None,
    description: Optional[str] = None,
):
    """Register one MCP resource handler.

    Args:
        func: The Python function that resolves the resource content.
        uri: Optional resource URI template. Defaults to None.
        name: Optional resource name. Defaults to None.
        description: Optional resource description. Defaults to None.
    """
    if uri:
        self.mcp.resource(uri)(func)
    else:
        self.mcp.resource()(func)

add_prompt

add_prompt(
    func: Callable,
    name: Optional[str] = None,
    description: Optional[str] = None,
)

Register one MCP prompt handler.

参数：

名称	类型	描述	默认
func	Callable	The Python function that formats the prompt.	必需
name	Optional[str]	Optional explicit prompt name. Defaults to None.	None
description	Optional[str]	Optional prompt description. Defaults to None.	None

源代码位于： jianmu/mcp/server.py

def add_prompt(self, func: Callable, name: Optional[str] = None, description: Optional[str] = None):
    """Register one MCP prompt handler.

    Args:
        func: The Python function that formats the prompt.
        name: Optional explicit prompt name. Defaults to None.
        description: Optional prompt description. Defaults to None.
    """
    if name or description:
        self.mcp.prompt(name=name, description=description)(func)
    else:
        self.mcp.prompt()(func)

run

run(transport: str = 'stdio', **kwargs: Any) -> None

Start serving the MCP server with FastMCP.

参数：

名称	类型	描述	默认
transport	str	Subprocess transport style, e.g., 'stdio' or 'sse'. Defaults to 'stdio'.	'stdio'
kwargs	Any	Extra parameters passed to FastMCP.run.	{}

源代码位于： jianmu/mcp/server.py

def run(self, transport: str = "stdio", **kwargs: Any) -> None:
    """Start serving the MCP server with FastMCP.

    Args:
        transport: Subprocess transport style, e.g., 'stdio' or 'sse'. Defaults to 'stdio'.
        kwargs: Extra parameters passed to FastMCP.run.
    """
    self.mcp.run(transport=transport, **kwargs)

get_info

get_info() -> Dict[str, Any]

Return a compact description of registered MCP capabilities.

返回：

类型	描述
Dict[str, Any]	A dictionary containing server name, description, and protocol info.

源代码位于： jianmu/mcp/server.py

def get_info(self) -> Dict[str, Any]:
    """Return a compact description of registered MCP capabilities.

    Returns:
        A dictionary containing server name, description, and protocol info.
    """
    return {
        "name": self.name,
        "description": self.description,
        "protocol": "MCP",
    }

MCPServerBuilder

MCPServerBuilder(
    name: str, description: Optional[str] = None
)

Chainable MCP server builder.

属性：

名称	类型	描述
server		The MCPServer instance being configured.

Create a chainable builder for one MCP server.

源代码位于： jianmu/mcp/server.py

def __init__(self, name: str, description: Optional[str] = None):
    """Create a chainable builder for one MCP server."""
    self.server = MCPServer(name, description)

with_tool

with_tool(
    func: Callable,
    name: Optional[str] = None,
    description: Optional[str] = None,
) -> MCPServerBuilder

Register one tool and return the builder.

参数：

名称	类型	描述	默认
func	Callable	The Python function that implements the tool.	必需
name	Optional[str]	Optional tool name override. Defaults to None.	None
description	Optional[str]	Optional tool description. Defaults to None.	None

返回：

类型	描述
MCPServerBuilder	The resulting 'MCPServerBuilder' value.

源代码位于： jianmu/mcp/server.py

def with_tool(self, func: Callable, name: Optional[str] = None, description: Optional[str] = None) -> "MCPServerBuilder":
    """Register one tool and return the builder.

    Args:
        func: The Python function that implements the tool.
        name: Optional tool name override. Defaults to None.
        description: Optional tool description. Defaults to None.

    Returns:
        The resulting `'MCPServerBuilder'` value.
    """
    self.server.add_tool(func, name, description)
    return self

with_resource

with_resource(
    func: Callable,
    uri: Optional[str] = None,
    name: Optional[str] = None,
    description: Optional[str] = None,
) -> MCPServerBuilder

Register one resource and return the builder.

参数：

名称	类型	描述	默认
func	Callable	The Python function that resolves the resource content.	必需
uri	Optional[str]	Optional resource URI. Defaults to None.	None
name	Optional[str]	Optional resource name. Defaults to None.	None
description	Optional[str]	Optional resource description. Defaults to None.	None

返回：

类型	描述
MCPServerBuilder	The resulting 'MCPServerBuilder' value.

源代码位于： jianmu/mcp/server.py

def with_resource(
    self,
    func: Callable,
    uri: Optional[str] = None,
    name: Optional[str] = None,
    description: Optional[str] = None,
) -> "MCPServerBuilder":
    """Register one resource and return the builder.

    Args:
        func: The Python function that resolves the resource content.
        uri: Optional resource URI. Defaults to None.
        name: Optional resource name. Defaults to None.
        description: Optional resource description. Defaults to None.

    Returns:
        The resulting `'MCPServerBuilder'` value.
    """
    self.server.add_resource(func, uri, name, description)
    return self

with_prompt

with_prompt(
    func: Callable,
    name: Optional[str] = None,
    description: Optional[str] = None,
) -> MCPServerBuilder

Register one prompt and return the builder.

参数：

名称	类型	描述	默认
func	Callable	The Python function that formats the prompt.	必需
name	Optional[str]	Optional prompt name. Defaults to None.	None
description	Optional[str]	Optional prompt description. Defaults to None.	None

返回：

类型	描述
MCPServerBuilder	The resulting 'MCPServerBuilder' value.

源代码位于： jianmu/mcp/server.py

def with_prompt(self, func: Callable, name: Optional[str] = None, description: Optional[str] = None) -> "MCPServerBuilder":
    """Register one prompt and return the builder.

    Args:
        func: The Python function that formats the prompt.
        name: Optional prompt name. Defaults to None.
        description: Optional prompt description. Defaults to None.

    Returns:
        The resulting `'MCPServerBuilder'` value.
    """
    self.server.add_prompt(func, name, description)
    return self

build

build() -> MCPServer

Return the configured MCPServer instance.

返回：

类型	描述
MCPServer	The built MCPServer instance.

源代码位于： jianmu/mcp/server.py

def build(self) -> MCPServer:
    """Return the configured ``MCPServer`` instance.

    Returns:
        The built MCPServer instance.
    """
    return self.server

run

run(transport: str = 'stdio', **kwargs: Any) -> None

Run the configured server immediately.

参数：

名称	类型	描述	默认
transport	str	Stdio or SSE transport name. Defaults to 'stdio'.	'stdio'
kwargs	Any	Additional run options.	{}

源代码位于： jianmu/mcp/server.py

def run(self, transport: str = "stdio", **kwargs: Any) -> None:
    """Run the configured server immediately.

    Args:
        transport: Stdio or SSE transport name. Defaults to 'stdio'.
        kwargs: Additional run options.
    """
    self.server.run(transport=transport, **kwargs)