一、http方式运行服务和客户端

1.1 服务端

# coding=utf-8
'''
@File    ：demo_basic.py.py
@Author  ：juxuan2
@Date    ：2025/12/8 09:20 
'''
from fastmcp import FastMCP
from typing import Literal, List
import httpx

# 1. 初始化 FastMCP 服务器（配置元数据 + FastAPI 特性）
mcp = FastMCP(
    name="demo_mcp_server",
    instructions="FastMCP 2.0 演示服务器：天气查询 + 计算器 + 城市资源",
    version="1.0.0"
)

# 2. 定义工具：天气查询（对接 Open-Meteo 公开 API）
@mcp.tool(
    name="get_weather_forecast",  # 自定义工具名（默认取函数名）
    description="查询指定城市7天天气预报"
)
def get_weather_forecast(
    city: str,
    units: Literal["celsius", "fahrenheit"] = "celsius"
) -> str:
    """
    查询指定城市的7天天气预报
    :param city: 城市名称（中文/英文）
    :param units: 温度单位（摄氏度/华氏度）
    :return: 格式化天气预报
    """
    # 模拟城市坐标映射（实际场景可对接地理编码 API）
    city_coords = {
        "北京": (39.9042, 116.4074),
        "上海": (31.2304, 121.4737),
        "广州": (23.1291, 113.2644),
        "New York": (40.7128, -74.0060)
    }
    if city not in city_coords:
        return f"错误：暂不支持{city}，支持城市：{list(city_coords.keys())}"

    lat, lon = city_coords[city]
    # 调用 Open-Meteo 天气 API
    try:
        response = httpx.get(
            "https://api.open-meteo.com/v1/forecast",
            params={
                "latitude": lat,
                "longitude": lon,
                "daily": "temperature_2m_max,temperature_2m_min",
                "temperature_unit": units,
                "timezone": "Asia/Shanghai" if city in ["北京", "上海", "广州"] else "America/New_York"
            }
        )
        response.raise_for_status()
        data = response.json()

        # 格式化结果
        forecast = [f"{city} 7天天气预报（{units}）："]
        for date, max_temp, min_temp in zip(
            data["daily"]["time"],
            data["daily"]["temperature_2m_max"],
            data["daily"]["temperature_2m_min"]
        ):
            forecast.append(f"{date}：最低 {min_temp}°，最高 {max_temp}°")
        return "\n".join(forecast)
    except Exception as e:
        return f"查询失败：{str(e)}"

# 3. 定义工具：简单计算器（支持加减乘除）
@mcp.tool()
def calculator(
    a: float,
    b: float,
    operation: Literal["add", "subtract", "multiply", "divide"]
) -> str:
    """
    数学计算器
    :param a: 第一个数值
    :param b: 第二个数值
    :param operation: 运算类型（add/减/subtract/乘/multiply/除/divide）
    :return: 运算结果
    """
    if operation == "add":
        result = a + b
    elif operation == "subtract":
        result = a - b
    elif operation == "multiply":
        result = a * b
    elif operation == "divide":
        if b == 0:
            return "错误：除数不能为0"
        result = a / b
    else:
        return "错误：不支持的运算类型"
    return f"{a} {operation} {b} = {result:.2f}"

# 4. 定义资源：向 LLM 提供结构化数据（支持的城市列表）
@mcp.resource(
    "resource://supported_cities",
    name="supported_cities",
    description="天气查询工具支持的城市列表",
    mime_type="application/json"  # 使用mime_type代替type参数
)
def get_supported_cities() -> List[str]:
    """返回天气服务支持的城市列表"""
    return ["北京", "上海", "广州", "New York"]

# 5. 定义提示模板：指导 LLM 正确调用工具
@mcp.prompt(
    name="weather_query_guide",
    description="LLM 调用天气工具的指导模板"
)
def weather_query_guide() -> str:
    return """
    协助用户查询天气的规则：
    1. 先通过 supported_cities 资源确认用户查询的城市是否支持；
    2. 若支持，调用 get_weather_forecast 工具，指定 city 和 units 参数；
    3. 若不支持，告知用户并列出支持的城市；
    4. 格式化返回结果，确保清晰易读。
    """

# 6. 运行服务器（支持热重载、指定端口/主机）
if __name__ == "__main__":
    mcp.run(
    transport="http",  # 使用HTTP传输
    host="0.0.0.0",    # 允许外部访问
    port=8000          # 端口
)

1.2 客户端

# coding=utf-8
'''
@File    ：demo_client.py
@Author  ：juxuan2
@Date    ：2025/12/8 09:32 
'''
import asyncio

import fastmcp

async def load_mcp():
    client = fastmcp.Client("http://127.0.0.1:8000/mcp")
    async with client:
        print(await client.list_tools())

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

二、stdio方式运行服务和客户端

2.1 服务端

from fastmcp import FastMCP

# 创建MCP实例
mcp = FastMCP(
    instructions="这是一个简单的天气查询服务",
)

# 注册资源
@mcp.resource("resource://supported_cities", mime_type="application/json")
async def supported_cities():
    return ["北京", "上海", "广州", "深圳", "成都", "杭州"]

# 注册工具
@mcp.tool(
    name="weather_check",
    description="查询指定城市的天气信息",
    output_schema={
        "type": "object",
        "properties": {
            "location": {
                "type": "string",
                "description": "查询的城市名称"
            },
            "temperature": {
                "type": "integer",
                "description": "当前温度，单位摄氏度"
            },
            "condition": {
                "type": "string",
                "description": "天气状况"
            }
        }
    }
)
async def weather_check(location: str) -> dict:
    # 模拟天气数据
    weather_data = {
        "北京": {"temperature": 22, "condition": "晴朗"},
        "上海": {"temperature": 25, "condition": "多云"},
        "广州": {"temperature": 28, "condition": "小雨"},
        "深圳": {"temperature": 29, "condition": "晴天"},
        "成都": {"temperature": 20, "condition": "阴天"},
        "杭州": {"temperature": 24, "condition": "多云"}
    }
    
    # 检查城市是否支持
    if location not in weather_data:
        return {
            "location": location,
            "temperature": None,
            "condition": "未知城市"
        }
    
    data = weather_data[location]
    return {
        "location": location,
        "temperature": data["temperature"],
        "condition": data["condition"]
    }

# 运行服务器（stdio模式是默认的，无需指定transport）
mcp.run()

2.2 客户端

import asyncio
import sys
import traceback
from pathlib import Path
from fastmcp import Client
from fastmcp.client.transports import PythonStdioTransport

async def main():
    # 通过stdio方式连接到Python脚本服务器
    # 这里的"demo_stdio_server.py"是要运行的服务器脚本路径
    print(f"使用Python解释器: {sys.executable}")
    print(f"当前工作目录: {sys.path[0]}")
    
    try:
        # 直接使用PythonStdioTransport来连接，这样可以更清楚地看到错误
        transport = PythonStdioTransport(
            script_path="demo_stdio_server.py",
            log_file=Path("stdio_server.log")
        )
        
        client = Client(transport)
        
        print("尝试连接到服务器...")
        # 使用上下文管理器建立连接
        async with client:
            print("已连接到服务器")
            
            # 调用工具
            print("调用weather_check工具...")
            result = await client.call_tool("weather_check", {
                "location": "北京"
            })
            print(f"天气查询结果: {result}")
            
            # 获取资源
            print("获取supported_cities资源...")
            resource = await client.read_resource("resource://supported_cities")
            print(f"支持的城市列表: {resource}")
            
            # 调用另一个工具
            print("调用weather_check工具(上海)...")
            result2 = await client.call_tool("weather_check", {
                "location": "上海"
            })
            print(f"天气查询结果: {result2}")
    except Exception as e:
        print(f"连接或调用失败: {e}")
        traceback.print_exc()

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

或者

import asyncio
from fastmcp import Client

async def main():
    """通过stdio方式连接FastMCP服务器并调用工具示例"""
    
    # 方式1：直接传入脚本路径，Client会自动推断为stdio传输
    client = Client("demo_stdio_server.py")
    
    # 方式2：显式指定stdio传输（更灵活）
    # from fastmcp.client.transports import PythonStdioTransport
    # transport = PythonStdioTransport(script_path="demo_stdio_server.py")
    # client = Client(transport)
    
    try:
        # 建立连接
        async with client:
            print("✅ 已通过stdio方式连接到服务器")
            
            # 调用工具
            print("\n📡 调用weather_check工具...")
            result = await client.call_tool("weather_check", {
                "location": "北京"
            })
            
            # 处理结果
            if not result.is_error:
                print(f"✅ 工具调用成功：")
                print(f"   城市：{result.structured_content['location']}")
                print(f"   温度：{result.structured_content['temperature']}°C")
                print(f"   天气：{result.structured_content['condition']}")
            
            # 读取资源
            print("\n📁 读取supported_cities资源...")
            resource = await client.read_resource("resource://supported_cities")
            print(f"✅ 资源获取成功：")
            print(f"   支持的城市：{resource[0].text}")
            
    except Exception as e:
        print(f"❌ 连接或调用失败: {e}")

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

三、sse方式运行服务和客户端

3.1 服务端

from fastmcp import FastMCP

# 创建MCP实例
mcp = FastMCP(
    instructions="这是一个支持SSE传输的天气查询服务",
)

# 注册资源
@mcp.resource("resource://supported_cities", mime_type="application/json")
async def supported_cities():
    return ["北京", "上海", "广州", "深圳", "成都", "杭州"]

# 注册工具
@mcp.tool(
    name="weather_check",
    description="查询指定城市的天气信息",
    output_schema={
        "type": "object",
        "properties": {
            "location": {
                "type": "string",
                "description": "查询的城市名称"
            },
            "temperature": {
                "type": "integer",
                "description": "当前温度，单位摄氏度"
            },
            "condition": {
                "type": "string",
                "description": "天气状况"
            }
        }
    }
) 
async def weather_check(location: str) -> dict:
    # 模拟天气数据
    weather_data = {
        "北京": {"temperature": 22, "condition": "晴朗"},
        "上海": {"temperature": 25, "condition": "多云"},
        "广州": {"temperature": 28, "condition": "小雨"},
        "深圳": {"temperature": 29, "condition": "晴天"},
        "成都": {"temperature": 20, "condition": "阴天"},
        "杭州": {"temperature": 24, "condition": "多云"}
    }
    
    # 检查城市是否支持
    if location not in weather_data:
        return {
            "location": location,
            "temperature": None,
            "condition": "未知城市"
        }
    
    data = weather_data[location]
    return {
        "location": location,
        "temperature": data["temperature"],
        "condition": data["condition"]
    }

# 运行服务器，使用SSE传输
mcp.run(
    transport="sse",
    host="0.0.0.0",
    port=8001
)

3.2 客户端

import asyncio
from fastmcp import Client

async def main():
    # 使用SSE方式连接FastMCP服务器
    print("=== 使用SSE方式连接FastMCP服务器 ===")
    try:
        # 服务器URL（注意必须包含/sse路径）
        sse_url = "http://127.0.0.1:8001/sse"
        
        # 创建客户端实例（通过URL中的/sse路径自动推断SSE传输）
        print("创建FastMCP客户端实例...")
        client = Client(sse_url)
        
        # 建立连接
        print("尝试连接到服务器...")
        async with client:
            print("✅ 客户端连接成功!")
            
            # 读取资源
            print("\n📁 读取支持的城市列表...")
            resource = await client.read_resource("resource://supported_cities")
            print(f"✅ 资源获取成功：")
            print(f"   支持的城市：{resource[0].text}")
            
            # 调用工具
            print("\n📡 调用weather_check工具（北京）...")
            result = await client.call_tool("weather_check", {
                "location": "北京"
            })
            
            # 处理结果
            if not result.is_error:
                print(f"✅ 工具调用成功：")
                print(f"   城市：{result.structured_content['location']}")
                print(f"   温度：{result.structured_content['temperature']}°C")
                print(f"   天气：{result.structured_content['condition']}")
            else:
                print(f"❌ 工具调用失败：{result.error_message}")
            
    except Exception as e:
        print(f"❌ 连接或调用失败: {e}")
        import traceback
        traceback.print_exc()

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

四、streamable-http方式运行服务和客户端

4.1 服务端

from fastmcp import FastMCP

# 创建FastMCP实例
mcp = FastMCP(
    name="StreamableHttpDemoServer"
)

# 注册资源示例
@mcp.resource("resource://supported_cities")
def get_supported_cities():
    """获取支持的城市列表"""
    return ["北京", "上海", "广州", "深圳", "成都", "杭州"]

# 注册工具示例
@mcp.tool
async def weather_check(city: str) -> dict:
    """获取指定城市的天气信息"""
    # 模拟天气数据
    weather_data = {
        "北京": {"temperature": 22, "weather": "晴朗"},
        "上海": {"temperature": 25, "weather": "多云"},
        "广州": {"temperature": 30, "weather": "阵雨"},
        "深圳": {"temperature": 29, "weather": "多云"},
        "成都": {"temperature": 19, "weather": "阴天"},
        "杭州": {"temperature": 23, "weather": "晴朗"}
    }
    
    if city in weather_data:
        data = weather_data[city]
        return {
            "city": city,
            "temperature": data["temperature"],
            "weather": data["weather"]
        }
    else:
        return {
            "city": city,
            "temperature": 0,
            "weather": "未知城市"
        }

# 注册另一个工具示例
@mcp.tool
async def time_check() -> dict:
    """获取当前时间"""
    import datetime
    now = datetime.datetime.now()
    return {
        "time": now.strftime("%Y-%m-%d %H:%M:%S"),
        "timestamp": now.timestamp()
    }

if __name__ == "__main__":
    # 使用streamable-http传输方式启动服务器
    mcp.run(
        transport="streamable-http",
        host="0.0.0.0",
        port=8002
    )

4.2 客户端

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

async def main():
    """使用streamable-http方式连接FastMCP服务器并演示功能"""
    
    # 服务器URL（streamable-http通常使用普通HTTP路径）
    server_url = "http://127.0.0.1:8002/mcp"
    
    try:
        print("=== 使用streamable-http方式连接FastMCP服务器 ===")
        
        # 方式1：直接传入URL，Client会自动处理传输方式
        # 注意：streamable-http不需要特殊的URL路径标识
        print("\n1. 创建客户端实例...")
        client = Client(server_url)
        
        # 方式2：显式创建StreamableHttpTransport
        # client = Client(StreamableHttpTransport(url=server_url))
        
        # 建立连接
        print("2. 尝试连接到服务器...")
        async with client:
            print("✅ 客户端连接成功!")
            
            # 获取所有可用工具
            print("\n3. 获取所有可用工具...")
            tools = await client.list_tools()
            
            print(f"✅ 发现 {len(tools)} 个可用工具:")
            for i, tool in enumerate(tools, 1):
                print(f"   {i}. {tool.name} - {tool.description}")
                
                # 打印工具的输入输出模式（如果有）
                if hasattr(tool, 'inputSchema') and tool.inputSchema:
                    print(f"      输入模式: {tool.inputSchema.type if hasattr(tool.inputSchema, 'type') else 'object'}")
                if hasattr(tool, 'outputSchema') and tool.outputSchema:
                    print(f"      输出模式: {tool.outputSchema.type if hasattr(tool.outputSchema, 'type') else 'object'}")
            
            # 读取资源示例
            print("\n4. 读取支持的城市列表...")
            resource_contents = await client.read_resource("resource://supported_cities")
            print(f"✅ 资源获取成功：")
            # 处理资源内容
            if hasattr(resource_contents, '__iter__') and not isinstance(resource_contents, str):
                # 如果是多个资源项
                city_list = []
                for item in resource_contents:
                    if hasattr(item, 'text'):
                        city_list.append(item.text)
                    else:
                        city_list.append(str(item))
                print(f"   支持的城市：{', '.join(city_list)}")
            elif hasattr(resource_contents, 'text'):
                # 如果是单个资源项
                print(f"   支持的城市：{resource_contents.text}")
            else:
                # 如果是直接的字符串或列表
                print(f"   支持的城市：{resource_contents}")
            
            # 调用天气查询工具
            print("\n5. 调用天气查询工具（北京）...")
            result = await client.call_tool("weather_check", {
                "city": "北京"
            })
            
            if not result.is_error:
                print(f"✅ 工具调用成功：")
                print(f"   城市：{result.structured_content['city']}")
                print(f"   温度：{result.structured_content['temperature']}°C")
                print(f"   天气：{result.structured_content['weather']}")
            else:
                print(f"❌ 工具调用失败：{result.error_message}")
            
            # 调用时间查询工具
            print("\n6. 调用时间查询工具...")
            result = await client.call_tool("time_check", {})
            
            if not result.is_error:
                print(f"✅ 工具调用成功：")
                print(f"   当前时间：{result.structured_content['time']}")
                print(f"   时间戳：{result.structured_content['timestamp']}")
            else:
                print(f"❌ 工具调用失败：{result.error_message}")
            
            print("\n🎉 所有操作完成!")
            
    except Exception as e:
        print(f"❌ 操作失败: {e}")
        import traceback
        traceback.print_exc()

# 运行主函数
if __name__ == "__main__":
    asyncio.run(main())