uv init mcp-project
cd mcp-project

uv add "mcp[cli]"

pip install "mcp[cli]"

uv run mcp

# server.py
from mcp.server.fastmcp import FastMCP

# 創建 MCP 服務器
mcp = FastMCP("Demo")


# 添加一個加法工具
@mcp.tool()
def add(a: int, b: int) -> int:
    """將兩個數字相加"""
    return a + b


# 添加一個動態問候資源
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """獲取個性化問候"""
    returnf"Hello, {name}!"

mcp dev server.py

import asyncio
import contextlib
from collections.abc import AsyncIterator

from mcp.server.lowlevel import Server
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from starlette.applications import Starlette
from starlette.routing import Mount
from starlette.types import Receive, Scope, Send

# 創建MCP服務器
app = Server("my-mcp-server")

# 定義工具和資源
# 例如: @app.call_tool(), @app.read_resource() 等

# 創建無狀態會話管理器
session_manager = StreamableHTTPSessionManager(
    app=app,
    event_store=None,  # 無狀態模式不需要事件存儲
    json_respnotallow=False,  # 使用SSE流而不JSON響應
    stateless=True,  # 啟用無狀態模式
)

# 處理Streamable HTTP請求
asyncdef handle_streamable_http(scope: Scope, receive: Receive, send: Send) -> None:
    await session_manager.handle_request(scope, receive, send)

# 定義生命周期管理
@contextlib.asynccontextmanager
asyncdef lifespan(app: Starlette) -> AsyncIterator[None]:
    asyncwith session_manager.run():
        yield

# 創建ASGI應用
starlette_app = Starlette(
    routes=[Mount("/mcp", app=handle_streamable_http)],
    lifespan=lifespan,
)

# 啟動服務器
import uvicorn
uvicorn.run(starlette_app, host="127.0.0.1", port=3000)

mkdir -p mcp-light-switch/mcp_light_switch
touch mcp-light-switch/pyproject.toml
touch mcp-light-switch/mcp_light_switch/__init__.py
touch mcp-light-switch/mcp_light_switch/server.py

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "mcp-light-switch"
version = "0.1.0"
description = "MCP Light Switch Server Example"
requires-python = ">=3.10"
dependencies = [
    "mcp>=0.1.0",
    "uvicorn>=0.24.0",
    "starlette>=0.31.0",
]

[project.scripts]
mcp-light-switch = "mcp_light_switch.server:main"

import asyncio
import contextlib
import logging
from collections.abc import AsyncIterator
from dataclasses import dataclass, field
from typing import Any, Dict, List

import click
import mcp.types as types
from mcp.server.lowlevel import Server
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from starlette.applications import Starlette
from starlette.routing import Mount
from starlette.types import Receive, Scope, Send

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)

# 電燈狀態類
@dataclass
class LightState:
    is_on: bool = False
    last_changed: str = "Never"# 最后一次狀態變更時間

# 全局電燈狀態
light_state = LightState()

def create_mcp_server() -> Server:
    """創建并配置MCP服務器"""
    app = Server("mcp-light-switch")

    @app.call_tool()
    asyncdef call_tool(name: str, arguments: Dict[str, Any]) -> List[types.TextContent]:
        """處理工具調用"""
        ctx = app.request_context
        result = ""

        # 根據工具名稱執行相應操作
        if name == "turn-on-light":
            result = turn_on_light()
        elif name == "turn-off-light":
            result = turn_off_light()
        else:
            raise ValueError(f"Unknown tool: {name}")

        # 發送日志消息
        await ctx.session.send_log_message(
            level="info",
            data=f"執行工具: {name}, 結果: {result}",
            logger="light_switch",
            related_request_id=ctx.request_id,
        )

        # 返回結果
        return [
            types.TextContent(
                type="text",
                text=result,
            )
        ]

    @app.list_tools()
    asyncdef list_tools() -> List[types.Tool]:
        """列出可用工具"""
        return [
            types.Tool(
                name="turn-on-light",
                descriptinotallow="打開電燈",
                inputSchema={
                    "type": "object",
                    "properties": {},
                },
            ),
            types.Tool(
                name="turn-off-light",
                descriptinotallow="關閉電燈",
                inputSchema={
                    "type": "object",
                    "properties": {},
                },
            ),
        ]

    @app.read_resource()
    asyncdef get_resource(uri: str) -> types.ReadResourceResult:
        """獲取資源"""
        # 打印接收到的URI以便調試
        logger.info(f"Received resource URI request: {uri}")
        
        # 使用標準URL格式
        if uri == "https://light.example/status":
            # 返回當前電燈狀態
            return types.ReadResourceResult(
                cnotallow=types.TextContent(
                    type="text",
                    text=get_light_status_text(),
                ),
                metadata={
                    "content-type": "text/plain",
                },
            )
        else:
            logger.error(f"Unknown resource URI: {uri}")
            raise ValueError(f"Unknown resource URI: {uri}")

    @app.list_resources()
    asyncdef list_resources() -> List[types.Resource]:
        """列出可用資源"""
        return [
            types.Resource(
                uri="light://status",
                descriptinotallow="當前電燈狀態",
                name="light_status",
            ),
        ]

    return app

def turn_on_light() -> str:
    """打開電燈"""
    global light_state
    if light_state.is_on:
        return"電燈已經是打開狀態"
    
    light_state.is_on = True
    light_state.last_changed = asyncio.get_event_loop().time()
    logger.info("電燈已打開")
    return"電燈已打開"

def turn_off_light() -> str:
    """關閉電燈"""
    global light_state
    ifnot light_state.is_on:
        return"電燈已經是關閉狀態"
    
    light_state.is_on = False
    light_state.last_changed = asyncio.get_event_loop().time()
    logger.info("電燈已關閉")
    return"電燈已關閉"

def get_light_status_text() -> str:
    """獲取電燈狀態文本描述"""
    global light_state
    status = "開啟"if light_state.is_on else"關閉"
    returnf"電燈狀態: {status}\n最后變更: {light_state.last_changed}"

@click.command()
@click.option("--port", default=3000, help="服務器監聽端口")
@click.option("--host", default="127.0.0.1", help="服務器監聽地址")
@click.option("--json-response", is_flag=True, help="使用JSON響應而不是SSE流")
def main(port: int, host: str, json_response: bool) -> int:
    """啟動MCP電燈開關服務器"""
    # 創建MCP服務器
    app = create_mcp_server()
    
    # 創建會話管理器
    session_manager = StreamableHTTPSessionManager(
        app=app,
        event_store=None,
        json_respnotallow=json_response,
        stateless=True,
    )
    
    # 處理Streamable HTTP請求
    asyncdef handle_streamable_http(scope: Scope, receive: Receive, send: Send) -> None:
        await session_manager.handle_request(scope, receive, send)
    
    # 定義生命周期管理
    @contextlib.asynccontextmanager
    asyncdef lifespan(app: Starlette) -> AsyncIterator[None]:
        asyncwith session_manager.run():
            logger.info(f"MCP電燈開關服務器已啟動! 訪問: http://{host}:{port}/mcp")
            yield
            logger.info("MCP電燈開關服務器已關閉")
    
    # 創建ASGI應用
    starlette_app = Starlette(
        debug=True,
        routes=[Mount("/mcp", app=handle_streamable_http)],
        lifespan=lifespan,
    )
    
    # 啟動服務器
    import uvicorn
    uvicorn.run(starlette_app, host=host, port=port)
    return0

if __name__ == "__main__":
    main()

pip install uv

cd mcp-light-switch
uv pip install -e .
uv run mcp_light_switch/server.py

npm install -g @modelcontextprotocol/inspector

npx @modelcontextprotocol/inspector

npx @modelcontextprotocol/inspector

from fastmcp import Client

# 創建客戶端，自動使用StreamableHttpTransport連接HTTP URL
client = Client("http://localhost:3000/mcp")

uv add fastmcp

uv pip install "httpx[socks]"

from fastmcp import Client
import asyncio

asyncdef main():
    # 創建客戶端并連接到電燈開關服務器
    client = Client("http://localhost:3000/mcp")
    
    asyncwith client:
        # 列出可用的工具
        tools = await client.list_tools()
        print(f"\n可用的工具: {tools}")
        
        # 列出可用的資源
        resources = await client.list_resources()
        print(f"\n可用的資源: {resources}")
        
        # 讀取電燈狀態
        status = await client.read_resource("light://status")
        print(f"\n當前電燈狀態: {status}")
        
        # 打開電燈
        print("\n打開電燈...")
        result = await client.call_tool("turn-on-light")
        print(f"結果: {result}")
        
        # 再次讀取電燈狀態
        status = await client.read_resource("light://status")
        print(f"\n電燈狀態現在是: {status}")
        
        # 關閉電燈
        print("\n關閉電燈...")
        result = await client.call_tool("turn-off-light")
        print(f"結果: {result}")
        
        # 再次讀取電燈狀態
        status = await client.read_resource("light://status")
        print(f"\n電燈狀態現在是: {status}")

# 運行客戶端
asyncio.run(main())

uv run python mcp_simple_client/client.py

可用的工具: ['turn-on-light', 'turn-off-light']

可用的資源: ['light://status']

當前電燈狀態: {'status': 'off'}

打開電燈...
結果: {'success': True}

電燈狀態現在是: {'status': 'on'}

關閉電燈...
結果: {'success': True}

電燈狀態現在是: {'status': 'off'}