FastMCP 入門 (2) - FastMCPサーバ

from fastmcp import FastMCP

# 基本的なサーバインスタンスを作成
mcp = FastMCP(name="MyAssistantServer")

# サーバとのやり取り方法に関する指示を追加することもできる
mcp_with_instructions = FastMCP(
    name="HelpfulAssistant",
    instructions="このサーバはデータ分析ツールを提供します。数値データを分析するには、get_average() を呼び出します。"
)

@mcp.tool()
def multiply(a: float, b: float) -> float:
    """2つの数値を掛け合わせる"""
    return a * b

@mcp.resource("data://config")
def get_config() -> dict:
    """アプリケーション構成を提供"""
    return {"theme": "dark", "version": "1.0"}

@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: int) -> dict:
    """IDでユーザープロファイルを取得"""
    # URI内の {user_id} が抽出され、この関数に渡される
    return {"id": user_id, "name": f"User {user_id}", "status": "active"}

@mcp.prompt()
def analyze_data(data_points: list[float]) -> str:
    """数値データの分析を求めるプロンプトを作成"""
    formatted_data = ", ".join(str(point) for point in data_points)
    return f"データポイントを分析してください: {formatted_data}"

# my_server.py
from fastmcp import FastMCP

mcp = FastMCP(name="MyServer")

@mcp.tool()
def greet(name: str) -> str:
    """ユーザーの名前であいさつ"""
    return f"Hello, {name}!"

if __name__ == "__main__":
    # このコードはファイルが直接実行された場合にのみ実行される
    
    # デフォルト設定での基本実行（stdioトランスポート）
    mcp.run()
    
    # 特定のトランスポートとパラメータの使用
    # mcp.run(transport="sse", host="127.0.0.1", port=9000)

# stdioで実行 (デフォルト)
mcp.run()  # または明示的に: mcp.run(transport="stdio")

# デフォルトのホスト/ポート (0.0.0.0:8000) でSSEを使用して実行
mcp.run(transport="sse")

# 特定のパラメータで設定
mcp.run(
    transport="sse", 
    host="127.0.0.1",  # Override default host
    port=8888,         # Override default port
    log_level="debug"  # Set logging level
)

# 同じパラメータで非同期的に実行
import asyncio
asyncio.run(
    mcp.run_sse_async(
        host="127.0.0.1", 
        port=8888, 
        log_level="debug"
    )
)

# SSEトランスポートの場合、kwargsはrun_sse_async()に渡される
mcp.run(transport="sse", **transport_kwargs)

# stdioトランスポートの場合、kwargsはrun_stdio_async()に渡される
mcp.run(transport="stdio", **transport_kwargs)

# サーバを実行 (デフォルトはstdioトランスポート)
fastmcp run my_server.py:mcp

# トランスポートを明示的に指定
fastmcp run my_server.py:mcp --transport sse

# ホストとポートでSSEトランスポートを構成
fastmcp run my_server.py:mcp --transport sse --host 127.0.0.1 --port 8888

# ログレベル付き
fastmcp run my_server.py:mcp --transport sse --log-level DEBUG

# 例: サブサーバのインポート
from fastmcp import FastMCP
import asyncio

main = FastMCP(name="Main")
sub = FastMCP(name="Sub")

@sub.tool()
def hello(): 
    return "hi"

main.mount("sub", sub)

from fastmcp import FastMCP, Client

backend = Client("http://example.com/mcp/sse")
proxy = FastMCP.from_client(backend, name="ProxyServer")
# 他のFastMCPサーバと同様にプロキシを使用

from fastmcp import FastMCP

# 初期化中に設定
mcp = FastMCP(
    name="ConfiguredServer",
    port=8080, # ServerSettingsに直接マップ
    on_duplicate_tools="error" # 重複処理を設定
)

# 設定はmcp.settingsからアクセスできる
print(mcp.settings.port) # 出力: 8080
print(mcp.settings.on_duplicate_tools) # 出力: "error"

from fastmcp import FastMCP
from mcp.server.auth.settings import (
    RevocationOptions,
    ClientRegistrationOptions,
    AuthSettings,
)


# 認証付きサーバを作成
mcp = FastMCP(
    name="SecureApp",
    auth_provider=MyOAuthServerProvider(),
    auth=AuthSettings(
        issuer_url="https://myapp.com",
        revocation_options=RevocationOptions(
            enabled=True,
        ),
        client_registration_options=ClientRegistrationOptions(
            enabled=True,
            valid_scopes=["myscope", "myotherscope"],
            default_scopes=["myscope"],
        ),
        required_scopes=["myscope"],
    ),
)