Open8

microsoft mcp-for-beginnersをなぞる

さみっとさみっと

https://github.com/microsoft/mcp-for-beginners/tree/main/03-GettingStarted/01-first-server

3-1 first-server

mcpのテスト
uv run python server.py
→ターミナルでの反応なし。普通にInspector使うことが多そうなのでスルー

Inspectorの起動コマンド
npx @modelcontextprotocol/inspector uv run server.py

ターミナル上に出てくるリンクを開くとブラウザで確認できる

さみっとさみっと

3-2 client

https://github.com/microsoft/mcp-for-beginners/tree/main/03-GettingStarted/02-client

最終的なコード

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

# Create server parameters for stdio connection
server_params = StdioServerParameters(
    command="uv",  # Executable
    args=["run", "python", "server.py"],  # Optional command line arguments
    env=None,  # Optional environment variables
)

async def run():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(
            read, write
        ) as session:
            # Initialize the connection
            await session.initialize()

            # List available resources
            resources = await session.list_resources()
            print("LISTING RESOURCES")
            for resource in resources:
                print("Resource: ", resource)

            # List available tools
            tools = await session.list_tools()
            print("LISTING TOOLS")
            for tool in tools.tools:
                print("Tool: ", tool.name)

            # Read a resource
            print("READING RESOURCE")
            content, mime_type = await session.read_resource("greeting://hello")

            # Call a tool
            print("CALL TOOL")
            result = await session.call_tool("add", arguments={"a": 1, "b": 7})
            print(result.content)



if __name__ == "__main__":
    import asyncio

    asyncio.run(run())

実行コマンド
uv run python client.py

さみっとさみっと

3-3 llm-client

https://github.com/microsoft/mcp-for-beginners/blob/main/03-GettingStarted/03-llm-client

3-2のclient.pyを編集していく

GitHubのトークン作成

  • GitHub TokenにてPersonal Access Tokenの権限でmodels権限だけ追加する必要がある

最終的なコード

  • LLM部分のロジックについて
    • LLM: ツールの選択・指示のみ(実行はしない)
    • MCPサーバー: 実際の処理・計算を実行
    • LLMは「秘書」、MCPサーバーは「実行者」の関係
  • tool_callsのイメージ
tool_calls = [
    {
        "id": "call_abc123",
        "function": {
            "name": "add",
            "arguments": '{"a": 2, "b": 20}'  # JSON文字列
        }
    }
]
  • トークンは.envファイルにGITHUB_TOKENとして登録
  • dotenvで呼び出し
import json
import os

from azure.ai.inference import ChatCompletionsClient
from azure.core.credentials import AzureKeyCredential
from dotenv import load_dotenv
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

dotenv_path = os.path.join(os.path.dirname(__file__), '.env')
load_dotenv(dotenv_path)

# Create server parameters for stdio connection
server_params = StdioServerParameters(
    command="uv",  # Executable
    args=["run", "python", "server.py"],  # Optional command line arguments
    env=None,  # Optional environment variables
)


def convert_to_llm_tool(tool):
    tool_schema = {
        "type": "function",
        "function": {
            "name": tool.name,
            "description": tool.description,
            "type": "function",
            "parameters": {
                "type": "object",
                "properties": tool.inputSchema["properties"]
            }
        }
    }

    return tool_schema


# llm

def call_llm(prompt, functions):
    token = os.environ["GITHUB_TOKEN"]
    endpoint = "https://models.inference.ai.azure.com"

    model_name = "gpt-4o"

    client = ChatCompletionsClient(
        endpoint=endpoint,
        credential=AzureKeyCredential(token),
    )

    print("CALLING LLM")
    response = client.complete(
        messages=[
            {
            "role": "system",
            "content": "You are a helpful assistant.",
            },
            {
            "role": "user",
            "content": prompt,
            },
        ],
        model=model_name,
        tools = functions,
        # Optional parameters
        temperature=1.,
        max_tokens=1000,
        top_p=1.    
    )

    response_message = response.choices[0].message
    
    functions_to_call = []

    if response_message.tool_calls:
        for tool_call in response_message.tool_calls:
            print("TOOL: ", tool_call)
            name = tool_call.function.name
            args = json.loads(tool_call.function.arguments)
            functions_to_call.append({ "name": name, "args": args })

    return functions_to_call


async def run():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(
            read, write
        ) as session:
            # Initialize the connection
            await session.initialize()

            # List available resources
            resources = await session.list_resources()
            print("LISTING RESOURCES")
            for resource in resources:
                print("Resource: ", resource)

            # List available tools
            tools = await session.list_tools()
            print("LISTING TOOLS")
            functions = []  # 関数リストを初期化
            for tool in tools.tools:
                print("Tool: ", tool.name)
                print("Tool", tool.inputSchema["properties"])
                functions.append(convert_to_llm_tool(tool))

            # Read a resource
            print("READING RESOURCE")
            content, mime_type = await session.read_resource("greeting://hello")

            # Call a tool
            print("CALL TOOL")
            result = await session.call_tool("add", arguments={"a": 1, "b": 7})
            print(result.content)
            
            prompt = "Add 2 to 20"

            # ask LLM what tools to all, if any
            functions_to_call = call_llm(prompt, functions)

            # call suggested functions
            for f in functions_to_call:
                result = await session.call_tool(f["name"], arguments=f["args"])
                print("TOOLS result: ", result.content)


if __name__ == "__main__":
    import asyncio

    asyncio.run(run())
さみっとさみっと

3-6 http-streaming

https://github.com/microsoft/mcp-for-beginners/tree/main/03-GettingStarted/06-http-streaming

トランスポートの種類とできること

転送方式 リアルタイム 更新 ストリーミング 拡張性 用途
stdio なし なし ローカルCLIツール
SSE あり あり Web、リアルタイム更新
ストリーミング対応HTTP あり あり クラウド、マルチクライアント

従来のHTTPストリーミングとMCP ストリーミングの違い

機能 従来のHTTPストリーミング MCPストリーミング(通知)
メインレスポンス チャンク形式 最後に単一レスポンス
進捗更新 データチャンクとして送信 通知として送信
クライアント要件 ストリーム処理必須 メッセージハンドラー実装必須
使用例 大容量ファイル、AIトークンストリーム 進捗、ログ、リアルタイムフィードバック
  • MCPにおけるストリーミングは、チャンク単位でのメインレスポンスの送信ではなく、ツールがリクエストを処理している間にクライアントに通知を送信すること

通知の例

{
  jsonrpc: "2.0";
  method: string;
  params?: {
    [key: string]: unknown;
  };
}

通知の実装

  • サーバーとクライアントの両方でリアルタイム更新を処理できる必要がある

実装

uv add "fastapi[all]" でライブラリ追加

あとは↓に従って試した
https://github.com/microsoft/mcp-for-beginners/blob/main/03-GettingStarted/06-http-streaming/solution/python/README.md

雰囲気は理解したが、バックエンド疎くて理解浅め
あとlogging周りの実装はちゃんと読めてない

さみっとさみっと

3-8 testing

https://github.com/microsoft/mcp-for-beginners/tree/main/03-GettingStarted/08-testing

  • mcp instpectorでのテスト
  • ユニットテスト:ここではpytestでlist_tools()をテストしている例があった

3-9 deployment

https://github.com/microsoft/mcp-for-beginners/tree/main/03-GettingStarted/09-deployment

  • デプロイ先
    • ローカル
    • コンテナ
    • クラウド
      • サーバーレス関数: 軽量な MCP サーバーをサーバーレス関数としてデプロイ
      • コンテナ サービス: Azure Container Apps、AWS ECS、Google Cloud Run など
      • Kubernetes :高可用性を実現するために、Kubernetes クラスターで MCP サーバーを展開および管理
さみっとさみっと

04 PracticalImplementation

https://github.com/microsoft/mcp-for-beginners/tree/main/04-PracticalImplementation

  • 手を動かす系ではなかった
  • C#、Java、TypeScript、JavaScript、PythonのMCP SDKでのサンプル実装例、Azure API ManagementでAPI管理をしながらMCPサーバーを扱う例、MCPサーバーのAzureへのデプロイ例が載っている
  • 割と03と内容被りな印象

雰囲気掴めたので一旦この辺でなぞるのストップ