# 假設(shè) mcp 已經(jīng)正確導(dǎo)入
try:
    from mcp import tool
except ImportError:
    # 如果 mcp 未找到，模擬一個(gè) tool 裝飾器
    def tool(func):
        return func

# 在 Service 業(yè)務(wù)代碼頭上添加 @tool 裝飾器
@tool
async def get_city_list(self) -> list:
    """獲取所有的城市信息。
    返回:
    str: 所有的城市信息列表
    """
    logging.info(f"獲取所有的城市信息")
    city_list = []
    for city in self.CITY_WEATHER_DATA:
        city_list.append(city)
    return city_list

from mcp.server.fastmcp import FastMCP
from city_01_service import CityDataServer
# 1-初始化 MCP 服務(wù)器
mcp = FastMCP("CityDataServer")
# 2-初始化城市信息服務(wù)器(業(yè)務(wù)代碼+@tool裝飾器)
city_server = CityDataServer()
# 3-在 Service 業(yè)務(wù)代碼頭上添加@mcp.tool()裝飾器
@mcp.tool()
# 獲取所有城市列表
async def get_city_list():
    """獲取所有城市列表。
    返回:
    str: 所有城市列表
    """
    city_list = await city_server.get_city_list()
    return city_list
# 4-主程序指定運(yùn)行方法-stdio進(jìn)程啟動(dòng)
if __name__ == "__main__":
    mcp.run(transport='stdio')

mkdir my_project
cd my_project
python -m venv .venv

# Windows
source .venv/Scripts/activate
# Mac
source .venv/bin/activate

# uv用于后續(xù)MCP Inspector的連接
pip install uv httpx mcp

import logging
# 假設(shè) mcp 已經(jīng)正確導(dǎo)入
try:
    from mcp import tool
except ImportError:
    # 如果 mcp 未找到，模擬一個(gè) tool 裝飾器
    def tool(func):
        return func
# 配置日志打印級(jí)別
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)
# 定義城市服務(wù)
class CityDataServer:
    # 模擬城市的天氣數(shù)據(jù)
    CITY_WEATHER_DATA = {
        "北京": {"condition": "晴", "temperature": 25, "humidity": 40},
        "上海": {"condition": "多云", "temperature": 27, "humidity": 60},
        "廣州": {"condition": "雨", "temperature": 30, "humidity": 80},
        "深圳": {"condition": "多云", "temperature": 29, "humidity": 70},
        "杭州": {"condition": "晴", "temperature": 26, "humidity": 50},
    }
    @tool
    async def get_city_weather(self, city: str) -> str:
        """獲取指定城市的天氣信息。
        參數(shù):
        city (str): 城市名稱(chēng)
        返回:
        str: 天氣信息描述
        """
        logging.info(f"獲取天氣信息: {city}")
        if city in self.CITY_WEATHER_DATA:
            weather = self.CITY_WEATHER_DATA[city]
            return f"{city} : {weather['condition']} , {weather['temperature']} °C,濕度 {weather['humidity']} %"
        else:
            return f"抱歉,未找到 {city} 的天氣信息"
    @tool
    async def get_city_list(self) -> list:
        """獲取所有的城市信息。
        返回:
        str: 所有的城市信息列表
        """
        logging.info(f"獲取所有的城市信息")
        city_list = []
        for city in self.CITY_WEATHER_DATA:
            city_list.append(city)
        return city_list
    @tool
    async def get_city_detail(self, city: str) -> str:
        """獲取指定城市的信息。
        參數(shù):
        city (str): 城市名稱(chēng)
        返回:
        str: 城市信息
        """
        logging.info(f"獲取指定城市的信息: {city}")
        if city in await self.get_city_list():
            return f"{city} : 一個(gè)風(fēng)景秀麗的城市，你值得去玩一把"
        else:
            return f"抱歉,未找到 {city} 的城市信息"

from mcp.server.fastmcp import FastMCP
from city_01_service import CityDataServer
# 初始化 MCP 服務(wù)器
mcp = FastMCP("CityDataServer")
# 初始化城市信息服務(wù)器
city_server = CityDataServer()
# 獲取天氣信息的工具
@mcp.tool()
async def get_city_weather(city: str) -> str:
    """獲取指定城市的天氣信息。
    參數(shù):
    city (str): 城市名稱(chēng)
    返回:
    str: 天氣信息描述
    """
    city_weather_info = await city_server.get_city_weather(city)
    return city_weather_info
@mcp.tool()
# 獲取所有城市列表
async def get_city_list():
    """獲取所有城市列表。
    返回:
    str: 所有城市列表
    """
    city_list = await city_server.get_city_list()
    return city_list
@mcp.tool()
# 獲取指定城市的信息
async def get_city_detail(city: str):
    """獲取指定城市的信息。
    參數(shù):
    city (str): 城市名稱(chēng)
    返回:
    str: 指定城市的信息
    """
    city_info = await city_server.get_city_detail(city)
    return city_info
# 主程序
if __name__ == "__main__":
    mcp.run(transport='stdio')

# 1-安裝MCP Inspector
pip install mcp[cli]

# 2-運(yùn)行MCP Inspector
mcp dev city_02_mcp_server.py