代码收藏家技术教程 2025-06-11

MCP基础知识 Python FastMCP MCP客户端开发+服务端开发

MCP是什么

MCP（Model Context Protocol，模型上下文协议）是由Anthropic提出的一项开放标准协议，旨在为大型语言模型（LLMs）与外部数据源及工具的交互建立统一接口。其核心目标是通过标准化通信方式，使AI模型能够安全、便捷地访问外部资源，例如数据库、API或特定工具，从而提升应用灵活性与扩展性。MCP的设计理念类似于“AI应用的USB-C接口”，通过模块化架构实现即插即用的集成模式，大幅降低开发复杂度。该协议自2024年底推出以来，已逐渐成为企业级AI应用开发的重要基础，尤其在跨系统数据融合与工具链扩展场景中展现出显著优势。相较于传统API，MCP通过统一的数据格式和交互规范，进一步简化了模型上下文的传递流程。

准备工作

在开始之前，请先安装Python，如果之前安装了fastmcp确保版本<2.0.0

安装FastMCP

pip instasll fastmcp

客户端

导入Fastmcp

import asyncio
from fastmcp import Client

选择一个Transport

FastMCP提供了多种的MCP Transport客户端，包括sse(Server-Sent Events)，StreamableHttp(流式HTTP)，Studio(标准输入/输出)

Client类接受一个Transport，可以使用的如下

# .venv\Lib\site-packages\fastmcp\client\__init__.py line 15-27
__all__ = [
    "Client", # 客户端
    "ClientTransport", # Transport的父类
    "WSTransport", 
    "SSETransport", # SSE的Transport
    "StdioTransport", # Stdio的Transport
    "PythonStdioTransport", # Python的Transport
    "NodeStdioTransport", # Nodejs的Transport
    "UvxStdioTransport",
    "NpxStdioTransport",
    "FastMCPTransport",
    "StreamableHttpTransport", # StreamableHttp的Transport
]

SSE

import asyncio
from fastmcp import Client
from fastmcp.client import SSETransport # 使用SSETransport 

client = Client(SSETransport("http://localhost:8000/sse"))

参数如下

class SSETransport(ClientTransport):
    """Transport implementation that connects to an MCP server via Server-Sent Events."""

    def __init__(
        self,
        url: str | AnyUrl,
        headers: dict[str, str] | None = None,
        sse_read_timeout: datetime.timedelta | float | int | None = None,
    ):
        ...

参数名	使用
url	MCP SSE服务器的
headers	自定义请求头
sse_read_timeout	连接TimeOut时间

Stdio

import asyncio
from fastmcp import Client
from fastmcp.client import StdioTransport

client = Client(StdioTransport("python" ,["mcp_server.py"])) # 假设mcp_server.py存在

参数如下

class StdioTransport(ClientTransport):
    """
    Base transport for connecting to an MCP server via subprocess with stdio.

    This is a base class that can be subclassed for specific command-based
    transports like Python, Node, Uvx, etc.
    """

    def __init__(
        self,
        command: str,
        args: list[str],
        env: dict[str, str] | None = None,
        cwd: str | None = None,
    ):
    ...

参数名	说明	示例
command	执行命令需运行的可执行命令	python, node, uvx
args	传递给命令的参数列表	[“main.py”],[“test.js”]
env	为子进程设置的环境变量	{“ENV_NAME”: “value”}
cwd	子进程的当前工作目录	/home/mcp_client

初始化

客户端异步运行，必须在块内使用。此上下文管理器处理建立连接、初始化 MCP 会话以及在退出时清理资源。async with

可以使用client.is_connected()判断是否连接

import asyncio
from fastmcp import Client
from fastmcp.client import SSETransport

client = Client(SSETransport("http://localhost:8000/sse"))

async def main():
    async with client:
        print(f"Client状态: {client.is_connected()}")
        # 在这里使用MCP
    print(f"Client状态: {client.is_connected()}")

if __name__ == "__main__":
    asyncio.run(main())

获取可用的工具

tools = await client.list_tools()

获取服务器上可用工具的列表

返回值：list[mcp.types.Tool]：一个包含 Tool 对象的列表。

import asyncio
from fastmcp import Client
from fastmcp.client import SSETransport

client = Client(SSETransport("http://localhost:8000/sse")) # Assumes my_mcp_server.py exists

async def main():
    # Connection is established here
    async with client:
        print(f"Client connected: {client.is_connected()}")

        # Make MCP calls within the context
        tools = await client.list_tools()
        print(f"Available tools: {tools}")

    # Connection is closed automatically here
    print(f"Client connected: {client.is_connected()}")

if __name__ == "__main__":
    asyncio.run(main())

输出：

Client connected: True
Available tools: [Tool(name='add', description='计算两个整数的和', inputSchema={'properties': {'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}, 'required': ['a', 'b'], 'title': 'addArguments', 'type': 'object'}, annotations=None), Tool(name='sub', description='计算两个整数的差', inputSchema={'properties': {'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}, 'required': ['a', 'b'], 'title': 'subArguments', 'type': 'object'}, annotations=None), Tool(name='mul', description='计算两个整数的积', inputSchema={'properties': {'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}, 'required': ['a', 'b'], 'title': 'mulArguments', 'type': 'object'}, annotations=None), Tool(name='div', description='计算两个整数的商', inputSchema={'properties': {'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}, 'required': ['a', 'b'], 'title': 'divArguments', 'type': 'object'}, annotations=None), Tool(name='mod', description='计算两个整数的余数', inputSchema={'properties': {'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}, 'required': ['a', 'b'], 'title': 'modArguments', 'type': 'object'}, annotations=None), Tool(name='pow', description='计算两个整数的幂', inputSchema={'properties': {'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}, 'required': ['a', 'b'], 'title': 'powArguments', 'type': 'object'}, annotations=None)]
Client connected: False

调用工具

result = await client.call_tool("add", {"a": 1, "b": 2})

在服务器上调用一个工具。与 call_tool_mcp 不同，如果工具调用导致错误，此方法会引发 ToolError 异常。

参数：
name（str）：要调用的工具名称。
arguments（dict[str, Any] | None，可选）：传递给工具的参数。默认为 None。

添加一点功能

简单的一个调用Demo

import asyncio
from fastmcp import Client
from fastmcp.client import SSETransport
sse_url = "http://localhost:8000/sse"
client = Client(SSETransport(sse_url))
print("MCP客户端Demo",f"SSE客户端连接地址:{sse_url}")
async def main():
    # Connection is established here
    async with client:
        print(f"Client connected: {client.is_connected()}")

        tools = await client.list_tools()
        for tool in tools:
            print("-"* 20)
            print(f"工具名称: {tool.name}")
            print(f"工具描述: {tool.description}")
            print(f"工具参数: {tool.inputSchema["required"]}")
            print("-"*20)
        use = input("请输入要使用的工具名称: ") # 输入工具名称
        tools = await client.list_tools()
        if any(tool.name == use for tool in tools): # 判断add工具是否存在
            for tool in tools:
                if tool.name == use:
                    required = tool.inputSchema["required"]
                    break
            input_required = {}
            for i in required:
                input_required[i] = input(f"请输入{i}的值: ")
                if tool.inputSchema["properties"][i]["type"] == "int":
                    input_required[i] = int(input_required[i])
                elif tool.inputSchema["properties"][i]["type"] == "boolean":
                    input_required[i] = bool(input_required[i])
                elif tool.inputSchema["properties"][i]["type"] == "number":
                    input_required[i] = float(input_required[i])
                elif tool.inputSchema["properties"][i]["type"] == "string":
                    pass
                else:
                    print(f"未知类型：{tool.inputSchema['properties'][i]['type']}")
            result = await client.call_tool(tool.name, input_required)
            print(f"返回结果: {result}")

    # Connection is closed automatically here
    print(f"Client状态: {client.is_connected()}")

if __name__ == "__main__":
    asyncio.run(main())

服务端

先写代码，下次再介绍。。。

from fastmcp import FastMCP

# 创建一个 FastMCP 服务器实例
mcp = FastMCP("Calculator", instructions="这是一个简单的计算器,包含了加法、减法、乘法、除法、取余、幂运算")

# 定义加法工具
@mcp.tool()
def add(a: int, b: int) -> int:
    """计算两个整数的和"""
    return a + b
@mcp.tool()
def sub(a: int, b: int) -> int:
    """计算两个整数的差"""
    return a -b

@mcp.tool()
def mul(a: int, b: int) -> int:
    """计算两个整数的积"""
    return a * b

@mcp.tool()
def div(a: int, b: int) -> float:
    """计算两个整数的商"""
    return a / b

@mcp.tool()
def mod(a: int, b: int) -> int:
    """计算两个整数的余数"""
    return a % b

@mcp.tool()
def pow(a: int, b: int) -> int:
    """计算两个整数的幂"""
    return a ** b
if __name__ == "__main__":
    # 启动服务器
    mcp.run(transport='sse')

作者：城城000

物联沃分享整理
物联沃-IOTWORD物联网 » MCP基础知识 Python FastMCP MCP客户端开发+服务端开发