from mcp.server import FastMCP

# 初始化 MCP 服務(wù)器，名稱用于客戶端識別
mcp = FastMCP("my-server")

# 注冊工具：同步函數(shù)示例（加法器）
@mcp.tool()
def add(a: int, b: int) -> int:
    """計算兩個數(shù)的和"""
    return a + b

# 注冊工具：異步函數(shù)示例（模擬API請求）
@mcp.tool()
asyncdef fetch_data(url: str) -> str:
    """從URL獲取數(shù)據(jù)"""
    import httpx
    asyncwith httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.text

if __name__ == "__main__":
    mcp.run(transport="stdio")  # 使用標(biāo)準(zhǔn)輸入輸出通信

mcp run server.py --transport=sse

from mcp.client.stdio import stdio_client
from mcp import ClientSession, StdioServerParameters, types
import asyncio

# Client會使用這里的配置來啟動本地MCP Server
server_params = StdioServerParameters(
    command="python3",
    args=["./mcp_server.py"],
    env=None
)


asyncdef main():
    asyncwith stdio_client(server_params) as (read, write):
        asyncwith ClientSession(
                read, write, sampling_callback=None
        ) as session:
            await session.initialize()
            print(await session.list_tools())
            print('\n正在調(diào)用工具...')
            result = await session.call_tool("add", {"a": 1, "b": 2})
            print(result.content)
   time.sleep(60)

asyncio.run(main())

import sqlite3
from contextlib import contextmanager

@mcp.resource()
@contextmanager
def db_connection():
    """數(shù)據(jù)庫連接資源（自動關(guān)閉）"""
    conn = sqlite3.connect("data.db")
    try:
        yield conn
    finally:
        conn.close()

@mcp.tool()
def query_users(query: str, conn: sqlite3.Connection = mcp.depends(db_connection)) -> list:
    """執(zhí)行SQL查詢"""
    return conn.execute(query).fetchall()

from fastmcp import FastMCP
import openai  # 官方OpenAI庫

mcp = FastMCP("openai-integration")

# 配置OpenAI API密鑰（實際應(yīng)從環(huán)境變量讀?。?openai.api_key = "sk-your-api-key"

@mcp.prompt()
def role_based_prompt(role: str):
    """動態(tài)Prompt：根據(jù)角色設(shè)定AI行為"""
    role_profiles = {
        "teacher": "你是一個嚴(yán)謹?shù)臄?shù)學(xué)老師，必須逐步解釋推導(dǎo)過程",
        "joker": "用幽默的段子和網(wǎng)絡(luò)流行語回答問題",
        "coder": "僅返回代碼，不要任何解釋"
    }
    return {"system": role_profiles.get(role, "默認助手模式")}

# 給工具綁定prompt
@mcp.tool(prompt=role_based_prompt)
def ask_question(question: str, role: str = "teacher") -> str:
    """實際調(diào)用OpenAI生成回答（非模擬）"""
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[
            {"role": "system", "content": role_based_prompt(role)["system"]},
            {"role": "user", "content": question}
        ],
        temperature=0.7if role == "joker"else0.3# 幽默模式需要更高隨機性
    )
    return response.choices[0].message["content"]  # 返回AI生成的文本

if __name__ == "__main__":
    mcp.run(transport="http", port=8000)  # 啟動HTTP服務(wù)