FastMCP 入門 (7) - FastMCPクライアント

import asyncio
from fastmcp import Client, FastMCP

# トランスポートの例 (トランスポートページを参照)
server_instance = FastMCP(name="TestServer") # インメモリサーバ
sse_url = "http://localhost:8000/sse"       # SSEサーバのURL
ws_url = "ws://localhost:9000"             # WebSocketサーバのURL
server_script = "my_mcp_server.py"         # Pythonサーバファイルへのパス

# クライアントはトランスポートの種類を自動的に推測
client_in_memory = Client(server_instance)
client_sse = Client(sse_url)
client_ws = Client(ws_url)
client_stdio = Client(server_script)

print(client_in_memory.transport)
print(client_sse.transport)
print(client_ws.transport)
print(client_stdio.transport)

# 予想される出力 (環境によってタイプが若干異なる場合がある)
# <FastMCP(server='TestServer')>
# <SSE(url='http://localhost:8000/sse')>
# <WebSocket(url='ws://localhost:9000')>
# <PythonStdioTransport(command='python', args=['/path/to/your/my_mcp_server.py'])>

import asyncio
from fastmcp import Client

client = Client("my_mcp_server.py") # Assumes my_mcp_server.py exists

async def main():
    # ここで接続が確立
    async with client:
        print(f"Client connected: {client.is_connected()}")

        # コンテキスト内でMCP呼び出しを行う
        tools = await client.list_tools()
        print(f"Available tools: {tools}")

        if any(tool.name == "greet" for tool in tools):
            result = await client.call_tool("greet", {"name": "World"})
            print(f"Greet result: {result}")

    # ここで接続は自動的にクローズ
    print(f"Client connected: {client.is_connected()}")

if __name__ == "__main__":
    asyncio.run(main())

tools = await client.list_tools()
# tools -> list[mcp.types.Tool]

result = await client.call_tool("add", {"a": 5, "b": 3})
# result -> list[mcp.types.TextContent | mcp.types.ImageContent | ...]
print(result[0].text) # Assuming TextContent, e.g., '8'

resources = await client.list_resources()
# resources -> list[mcp.types.Resource]

templates = await client.list_resource_templates()
# templates -> list[mcp.types.ResourceTemplate]

# 静的リソースを読み取る
readme_content = await client.read_resource("file:///path/to/README.md")
# readme_content -> list[mcp.types.TextResourceContents | mcp.types.BlobResourceContents]
print(readme_content[0].text) # テキストを想定

# テンプレートから生成されたリソースを読み取る
weather_content = await client.read_resource("data://weather/london")
print(weather_content[0].text) # テキストJSONを想定

# 標準メソッド - ツールのリストのみを返します
tools = await client.list_tools()
# tools -> list[mcp.types.Tool]

# 生のMCPメソッド - 完全なプロトコルオブジェクトを返す
result = await client.list_tools_mcp()
# result -> mcp.types.ListToolsResult
tools = result.tools

import marvin
from fastmcp import Client
from fastmcp.client.sampling import (
    SamplingMessage,
    SamplingParams,
    RequestContext,
)

async def sampling_handler(
    messages: list[SamplingMessage],
    params: SamplingParams,
    context: RequestContext
) -> str:
    return await marvin.say_async(
        message=[m.content.text for m in messages],
        instructions=params.systemPrompt,
    )

client = Client(
    ...,
    sampling_handler=sampling_handler,
)

from fastmcp import Client
from fastmcp.client.logging import LogHandler, LogMessage

async def my_log_handler(params: LogMessage):
    print(f"[Server Log - {params.level.upper()}] {params.logger or 'default'}: {params.data}")

client_with_logging = Client(
    ...,
    log_handler=my_log_handler,
)

from fastmcp import Client

client = Client(
    ..., 
    roots=["/path/to/root1", "/path/to/root2"],
)

async def check_connection():
    async with client:
        await client.ping()
        print("Server is reachable")

async def safe_call_tool():
    async with client:
        try:
            # 「divide」ツールが存在すると仮定し、ZeroDivisionError が発生する可能性がある
            result = await client.call_tool("divide", {"a": 10, "b": 0})
            print(f"Result: {result}")
        except ClientError as e:
            print(f"Tool call failed: {e}")
        except ConnectionError as e:
            print(f"Connection failed: {e}")
        except Exception as e:
            print(f"An unexpected error occurred: {e}")

# ゼロ除算が発生した場合の出力例:
# ツールの呼び出しに失敗しました: ゼロ除算は許可されていません