fastmcp是一个非常好的mcp库，支持server端和client端。

但是google gemini对这个库的支持非常差。

生成的代码总是使用mcp这个库。用起来非常别扭。

这里把stdio,sse,http streamable三种模式都写了。

客户端主要的区别只是里面的transport不同.

transport = SSETransport(
    url=MCP_SERVER_URL,
    headers={"Authorization": "Bearer token"}
)
client = Client(transport)

主要参考：

https://gofastmcp.com/clients/transports

先是mcpserver.py

运行的时候，默认是http模式，通过参数指定哪种模式：

http, sse,stdio

运行前需要pip install fastmcp，以及其他的库

from fastmcp import FastMCP
import aiofiles
import hashlib



mcp = FastMCP("My MCP Server")
@mcp.tool
def send_email(address: str, content: str) -> str:
    '''Send a email to some address.'''
    print(f"send email to {address}, content: {content}")
    return f"Send email to {address} sucess!"



@mcp.tool
def greet(name: str) -> str:
    '''Send hello to my friend.'''
    return f"Hello, {name}!"

@mcp.tool
def add(a: int, b: int) -> int:
    '''Add two integer numbers together.'''
    return a + b


@mcp.tool
def special_add(a: int, b: int) -> int:
    '''一种特殊的加法, Add two integer numbers together and mod a special number.'''
    return (a + b)%8


@mcp.tool
def special_multi(a: int, b: int) -> int:
    '''一种特殊的乘法, multi two integer numbers together and mod a special number.'''
    return (a * b)%10


@mcp.tool
def md5sum(text: str) -> str:
    '''计算对应的字符串的md5 sum值.'''
    md5_hash = hashlib.md5()
    # 2. 将字符串转换为字节串并更新哈希对象
    md5_hash.update(text.encode('utf-8'))
    
    # 3. 获取并返回十六进制的哈希值
    return md5_hash.hexdigest()
    #return (a * b)%10



# Basic dynamic resource returning a string
@mcp.resource("resource://greeting")
def get_greeting() -> str:
    """Provides a simple greeting message."""
    return "Hello from FastMCP Resources!"





# Resource returning JSON data (dict is auto-serialized)
@mcp.resource("data://config")
def get_config() -> dict:
    """Provides application configuration as JSON."""
    return {
        "theme": "dark",
        "version": "1.2.0",
        "features": ["tools", "resources"],
    }
@mcp.resource("file:///data/log.txt", mime_type="text/plain")
async def read_log() -> str:
    """Reads content from a specific log file asynchronously."""
    try:
        async with aiofiles.open("./log.txt", mode="r") as f:
            content = await f.read()
        return content
    except FileNotFoundError:
        return "Log file not found."




# Basic prompt returning a string (converted to user message automatically)
@mcp.prompt
def ask_about_topic(topic: str) -> str:
    """Generates a user message asking for an explanation of a topic."""
    return f"Can you please explain the concept of '{topic}'?"








if __name__ == "__main__":
    print("不同的运行模式:\n")
    print("http,stdio,sse\n")
    print("default http mode.")
    import sys

    # 根据命令行参数选择运行模式，默认为 http
    mode = sys.argv[1] if len(sys.argv) > 1 else "http"

    if mode == "stdio":
        mcp.run()
    elif mode == "sse":
        mcp.run(transport="sse", host="0.0.0.0", port=8005)
    else:
        mcp.run(transport="http", host="0.0.0.0", port=8005)

主要的区别是客户端

http streamable

http streamable模式的客户端

import asyncio
from fastmcp.client.transports import StreamableHttpTransport
from fastmcp import Client

import os

TARGET_TOOL_NAME = "send_email"
TOOL_ARGUMENTS = {"address": "alice@gmail.com", "content": "Hello! can you give me 1000 USD?"}

MCP_SERVER_URL = os.getenv("MCP_SERVER_URL", "http://localhost:8005/mcp")


# Custom headers to send with the streamable connection
HTTP_HEADERS = {
    "Authorization": "Bearer your-token-here",
    "X-Custom-Header": "value",
}

transport = StreamableHttpTransport(
    url=MCP_SERVER_URL, 
    headers=HTTP_HEADERS
)
client = Client(transport)



async def run_fastmcp_custom_client():
    # 使用 fastmcp 的 ClientSession 进行交互
    async with client:
        try:
            # --- 功能 1: 列出工具 ---
            print("--- [远程工具列表] ---")
            tools_result = await client.list_tools()
            # FastMCP 的 session.list_tools() 返回一个包含 tools 列表的对象
            for tool in tools_result:
                print(f"🛠️  发现工具: {tool}")
            
            # --- 功能 2: 调用工具 ---
            TARGET_TOOL_NAME = "send_email"
            TOOL_ARGUMENTS = {"address": "alice@gmail.com", "content": "Hello! can you give me 1000 USD?"}
            
            print(f"\n🚀 正在调用工具")
            result = await client.call_tool(name=TARGET_TOOL_NAME, arguments=TOOL_ARGUMENTS)
            
            # 解析返回内容
            for content in result.content:
                if content.type == "text":
                    print(f"💡 执行结果: {content.text}")

        except Exception as e:
            print(f"❌ 运行中出错: {e}")

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

stdio 模式的客户端

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

import os

TARGET_TOOL_NAME = "send_email"
TOOL_ARGUMENTS = {"address": "alice@gmail.com", "content": "Hello! can you give me 1000 USD?"}

MCP_SERVER_URL = os.getenv("MCP_SERVER_URL", "http://localhost:8005/mcp")


# Custom headers to send with the streamable connection
HTTP_HEADERS = {
    "Authorization": "Bearer your-token-here",
    "X-Custom-Header": "value",
}


transport = StdioTransport(
    command="python",
    args=["mcpserver.py", "stdio"],
    env={"API_KEY": "secret", "LOG_LEVEL": "DEBUG"},
    cwd="./"
)
client = Client(transport)





async def run_fastmcp_custom_client():
    # 3. 使用 fastmcp 的 ClientSession 进行交互
    async with client:
        try:
            # --- 功能 1: 列出工具 ---
            print("--- [远程工具列表] ---")
            tools_result = await client.list_tools()
            # FastMCP 的 session.list_tools() 返回一个包含 tools 列表的对象
            for tool in tools_result:
                print(f"🛠️  发现工具: {tool}")
            
            # --- 功能 2: 调用工具 ---
            TARGET_TOOL_NAME = "send_email"
            TOOL_ARGUMENTS = {"address": "alice@gmail.com", "content": "Hello! can you give me 1000 USD?"}
            
            print(f"\n🚀 正在调用工具")
            result = await client.call_tool(name=TARGET_TOOL_NAME, arguments=TOOL_ARGUMENTS)
            
            # 解析返回内容
            for content in result.content:
                if content.type == "text":
                    print(f"💡 执行结果: {content.text}")

        except Exception as e:
            print(f"❌ 运行中出错: {e}")

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

sse模式的客户端代码

这次我们用的是mcp这个库。

import asyncio
from fastmcp import Client
from fastmcp.client.transports import StdioTransport
from fastmcp.client.transports import StreamableHttpTransport
from fastmcp.client.transports import SSETransport

import os

TARGET_TOOL_NAME = "send_email"
TOOL_ARGUMENTS = {"address": "alice@gmail.com", "content": "Hello! can you give me 1000 USD?"}

MCP_SERVER_URL = os.getenv("MCP_SERVER_URL", "http://localhost:8005/sse")


# Custom headers to send with the streamable connection
HTTP_HEADERS = {
    "Authorization": "Bearer your-token-here",
    "X-Custom-Header": "value",

}




transport = SSETransport(
    url=MCP_SERVER_URL,
    headers={"Authorization": "Bearer token"}
)
client = Client(transport)





async def run_fastmcp_custom_client():
    # 使用 fastmcp 的 ClientSession 进行交互
    async with client:
        try:
            # --- 功能 1: 列出工具 ---
            print("--- [远程工具列表] ---")
            tools_result = await client.list_tools()
            # FastMCP 的 session.list_tools() 返回一个包含 tools 列表的对象
            for tool in tools_result:
                print(f"🛠️  发现工具: {tool}")
            
            # --- 功能 2: 调用工具 ---
            TARGET_TOOL_NAME = "send_email"
            TOOL_ARGUMENTS = {"address": "alice@gmail.com", "content": "Hello! can you give me 1000 USD?"}
            
            print(f"\n🚀 正在调用工具")
            result = await client.call_tool(name=TARGET_TOOL_NAME, arguments=TOOL_ARGUMENTS)
            
            # 解析返回内容
            for content in result.content:
                if content.type == "text":
                    print(f"💡 执行结果: {content.text}")

        except Exception as e:
            print(f"❌ 运行中出错: {e}")

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