# 使用uv工具快速初始化項(xiàng)目（推薦）
uv init mcp_weather_server && cd mcp_weather_server
uv venv && source .venv/bin/activate  # Linux/Mac
# 或 .venv\Scripts\activate (Windows)

# 基礎(chǔ)版（適合個(gè)人開發(fā)）
uv add mcp>=1.9.1 requests python-dotenv
 
# 企業(yè)版（需處理高并發(fā)）
uv add fastmcp httpx uvicorn[standard]

# 創(chuàng)建test_install.py
from mcp.server.fastmcp import FastMCP
print("MCP SDK安裝成功" if 'FastMCP' in globals() else "安裝失敗")

# weather_tool.py
import requests
from mcp.server.fastmcp import tool
from dotenv import load_dotenv
import os
 
load_dotenv()  # 加載環(huán)境變量
 
@tool()
def get_weather(city: str) -> dict:
    """獲取城市實(shí)時(shí)天氣（需配置API_KEY）
    
    Args:
        city: 城市名稱（如"北京"）
    
    Returns:
        包含溫度、濕度、天氣狀況的字典
    """
    api_key = os.getenv('WEATHER_API_KEY')
    url = f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}&units=metric"
    
    try:
        response = requests.get(url, timeout=5)
        data = response.json()
        return {
            "temperature": data["main"]["temp"],
            "humidity": data["main"]["humidity"],
            "condition": data["weather"][0]["description"]
        }
    except Exception as e:
        return {"error": f"天氣查詢失敗: {str(e)}"}

# db_tool.py
import sqlite3
from mcp.server.fastmcp import tool
from mcp.types import TableContent
 
@tool()
def query_sales_data(limit: int = 10) -> TableContent:
    """查詢銷售數(shù)據(jù)表（示例用）
    
    Args:
        limit: 返回記錄數(shù)上限
    
    Returns:
        表格數(shù)據(jù)（包含列名和行數(shù)據(jù)）
    """
    conn = sqlite3.connect("sales.db")
    cursor = conn.cursor()
    cursor.execute(f"SELECT * FROM orders LIMIT {limit}")
    
    columns = [desc[0] for desc in cursor.description]
    rows = cursor.fetchall()
    
    return TableContent(
        type="table",
        columns=columns,
        rows=rows
    )

# server_stdio.py
from mcp.server.fastmcp import FastMCP
from weather_tool import get_weather
from db_tool import query_sales_data
 
mcp = FastMCP("Local Tools Server")
mcp.add_tool(get_weather)
mcp.add_tool(query_sales_data)
 
if __name__ == "__main__":
    mcp.run(mode="stdio")  # 通過標(biāo)準(zhǔn)輸入輸出通信

# server_sse.py
from fastmcp import FastMCP
from weather_tool import get_weather
import uvicorn
 
mcp = FastMCP("Cloud Tools Server")
mcp.add_tool(get_weather)
 
if __name__ == "__main__":
    # 通過HTTP事件流通信
    uvicorn.run(
        mcp.as_app(),  # 轉(zhuǎn)換為FastAPI應(yīng)用
        host="0.0.0.0",
        port=8000,
        workers=4  # 多進(jìn)程處理并發(fā)請求
    )

# converter.py
import json
from jsonschema import validate
 
def convert_to_mcp_tool(api_spec):
    """將OpenAPI接口轉(zhuǎn)換為MCP工具
    
    Args:
        api_spec: OpenAPI規(guī)范字典
    
    Returns:
        MCP工具代碼模板字符串
    """
    tool_template = f"""
@tool()
def {api_spec["operationId"]}(**kwargs):
    """
    {api_spec["summary"]}
    
    Parameters:
        {json.dumps(api_spec["parameters"], indent=4)}
    """
    # 這里添加實(shí)際調(diào)用邏輯
    return盛{{}}
    """
    return tool_template
 
# 示例：轉(zhuǎn)換天氣API
weather_api = {
    "operationId": "get_weather",
    "summary": "獲取城市天氣信息",
    "parameters": [
        {"name": "city", "in": "query", "required": True, "type": "string"}
    ]
}
print(convert_to_mcp_tool(weather_api))

{
  "mcpServers": {
    "weather_service": {
      "command": "python",
      "args": ["server_stdio.py", "--mode", "stdio"]
    }
  }
}

{
  "toolName": "get_weather",
  "arguments": {"city": "北京"}
}

{
  "temperature": 28,
  "humidity": 65,
  "condition": "多云"
}

npx @modelcontextprotocol/inspector python server_stdio.py

# 使用Vault管理密鑰
from hvac import Client
 
def get_secret(path: str) -> str:
    client = Client(url="http://vault-server:8200")
    return client.secrets.kv.v2.read_secret_version(path=path)["data"]["data"]["key"]

from pydantic import BaseModel, conint
 
class QueryParams(BaseModel):
    limit: conint(ge=1, le=100)  # 限制1-100的整數(shù)
 
@tool()
def safe_query(params: QueryParams) -> dict:
    validated = params.dict()  # 自動(dòng)校驗(yàn)參數(shù)
    # 執(zhí)行查詢邏輯...

from functools import partial
import asyncio
 
async def timed_fetch(url: str, timeout: float = 5.0):
    try:
        return await asyncio.wait_for(requests.get(url), timeout=timeout)
    except asyncio.TimeoutError:
        return {"error": "請求超時(shí)"}

from functools import lru_cache
 
@lru_cache(maxsize=100)
@tool()
def cached_weather(city: str) -> dict:
    # 首次調(diào)用執(zhí)行實(shí)際查詢
    # 后續(xù)調(diào)用直接返回緩存結(jié)果
    return _real_weather_query(city)

@tool()
async def async_data_processing(file_url: str):
    async with httpx.AsyncClient() as client:
        file_content = await client.get(file_url)
        # 并行處理數(shù)據(jù)...

# 使用Gunicorn部署FastMCP應(yīng)用
gunicorn server_sse:app -w 8 -k uvicorn.workers.UvicornWorker

# tools/stock_api.py
import httpx
from mcp.server.fastmcp import tool
from dotenv import load_dotenv
import os
 
load_dotenv()
 
@tool()
def get_stock_price(symbol: str) -> dict:
    """獲取股票實(shí)時(shí)價(jià)格（Alpha Vantage API）
    
    Args:
        symbol: 股票代碼（如"AAPL"）
    
    Returns:
        包含價(jià)格、漲跌幅的字典
    """
    api_key = os.getenv('ALPHA_VANTAGE_KEY')
    url = f"https://www.alphavantage.co/query?function=GLOBAL_QUOTE&symbol={symbol}&apikey={api_key}"
    
    response = httpx.get(url)
    data = response.json()
    
    if "Global Quote" not in data:
        return {"error": "未找到股票數(shù)據(jù)"}
    
    quote = data["Global Quote"]
    return {
        "symbol": symbol,
        "price": float(quote["05. price"]),
        "change": float(quote["09. change"]),
        "change_percent": float(quote["10. change percent"].strip('%'))
    }
 
# server.py
from fastmcp import FastMCP
from tools.stock_api import get_stock_price
from tools.data_analysis import calculate_stats
 
mcp = FastMCP("Stock Analysis Server")
mcp.add_tool(get_stock_price)
mcp.add_tool(calculate_stats)
 
if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

# 安裝依賴
uv install -r requirements.txt
 
# 啟動(dòng)服務(wù)器
python server.py
 
# 測試工具調(diào)用
curl -X POST http://localhost:8000/mcp/invoke \
  -H "Content-Type: application/json" \
  -d '{"toolName": "get_stock_price", "arguments": {"symbol": "AAPL"}}'