FastMCP 入門 (8) - クライアントトランスポート

from fastmcp import Client
from fastmcp.client.transports import PythonStdioTransport

server_script = "my_mcp_server.py" # Assumes this file exists and runs mcp.run()

# オプション1: 推定トランスポート
client_inferred = Client(server_script)

# オプション2: 明示的なトランスポート (特定のPython実行ファイルを使用する、または引数を追加するなど)
transport_explicit = PythonStdioTransport(
    script_path=server_script,
    python_cmd="/usr/bin/python3.11", # Pythonのバージョンを指定
    # args=["--some-server-arg"], # 引数を指定
    # env={"MY_VAR": "value"},   # 環境変数を指定
    # cwd="/path/to/run/in"       # 作業ディレクトリを指定
)
client_explicit = Client(transport_explicit)

async def use_stdio_client(client):
    async with client:
        tools = await client.list_tools()
        print(f"Connected via Python Stdio, found tools: {tools}")

# asyncio.run(use_stdio_client(client_inferred))
# asyncio.run(use_stdio_client(client_explicit))

from fastmcp import Client
from fastmcp.client.transports import NodeStdioTransport

node_server_script = "my_mcp_server.js" # Assumes this JS file starts an MCP server on stdio

# オプション1: 推定トランスポート
client_inferred = Client(node_server_script)

# オプション2: 明示的なトランスポート
transport_explicit = NodeStdioTransport(
    script_path=node_server_script,
    node_cmd="node" # またはNode実行ファイルへのパスを指定
)
client_explicit = Client(transport_explicit)

# 使い方は他のクライアントと同じ
# client_explicit を使用した非同期処理:
#    tools = await client_explicit.list_tools()

from fastmcp.client.transports import UvxStdioTransport

# 例: 仮想ツール「cloud-analyzer-mcp」を uvx経由で実行
# このツールを実行すると、stdio上で MCPサーバが起動すると仮定します
transport = UvxStdioTransport(
    tool_name="cloud-analyzer-mcp",
    # from_package="cloud-analyzer-cli", # ツール名が異なる場合はパッケージを任意で指定
    # with_packages=["boto3", "requests"], # 必要に応じて依存関係を追加
    # tool_args=["--config", "prod.yaml"] # ツール自体に引数を渡す
)
client = Client(transport)

# クライアントと非同期:
#     analysis = await client.call_tool("analyze_bucket", {"name": "my-data"})

from fastmcp.client.transports import NpxStdioTransport

# 例: npx経由で仮想的な「npm-mcp-server-package」を実行
transport = NpxStdioTransport(
    package="npm-mcp-server-package",
    # args=["--port", "stdio"] # パッケージスクリプトに渡される引数
)
client = Client(transport)

# クライアントと非同期:
#     response = await client.call_tool("get_npm_data", {})

from fastmcp import Client
from fastmcp.client.transports import SSETransport

sse_url = "http://localhost:8000/sse"

# オプション1: 推定トランスポート
client_inferred = Client(sse_url)

# オプション2: 明示的なトランスポート（カスタムヘッダーの追加など）
headers = {"Authorization": "Bearer mytoken"}
transport_explicit = SSETransport(url=sse_url, headers=headers)
client_explicit = Client(transport_explicit)

async def use_sse_client(client):
    async with client:
        tools = await client.list_tools()
        print(f"Connected via SSE, found tools: {tools}")

# asyncio.run(use_sse_client(client_inferred))
# asyncio.run(use_sse_client(client_explicit))

from fastmcp import Client
from fastmcp.client.transports import WSTransport

ws_url = "ws://localhost:9000"

# オプション1: 推定トランスポート
client_inferred = Client(ws_url)

# オプション2: 明示的なトランスポート
transport_explicit = WSTransport(url=ws_url)
client_explicit = Client(transport_explicit)

async def use_ws_client(client):
    async with client:
        tools = await client.list_tools()
        print(f"Connected via WebSocket, found tools: {tools}")

# asyncio.run(use_ws_client(client_inferred))
# asyncio.run(use_ws_client(client_explicit))

from fastmcp import FastMCP, Client
from fastmcp.client.transports import FastMCPTransport

# 1. FastMCPサーバーインスタンスを作成
server = FastMCP(name="InMemoryServer")
@server.tool()
def ping(): return "pong"

# 2. サーバインスタンスを直接参照するクライアントを作成
# オプションA: 推定
client_inferred = Client(server)

# オプションB: 明示的
transport_explicit = FastMCPTransport(mcp=server)
client_explicit = Client(transport_explicit)

# 3. クライアントを使用する (サブプロセスやネットワークは使用しない)
async def test_in_memory():
    async with client_inferred: # またはclient_explicit
        result = await client_inferred.call_tool("ping")
        print(f"In-memory call result: {result[0].text}") # Output: pong

# asyncio.run(test_in_memory())