MCP : クイックスタート : クライアント開発者向け (Python SDK)

すべての MCP サーバと統合可能な独自クライアントの構築を目指します。このチュートリアルでは、MCP サーバに接続する、LLP-powered チャットボット・クライアントを構築する方法を学習します。

Model Context Protocol (MCP) : クイックスタート : クライアント開発者向け (Python SDK)

作成 : クラスキャット・セールスインフォメーション
作成日時 : 05/07/2025

* 本記事は github modelcontextprotocol の以下のページを独自に翻訳した上でまとめ直し、補足説明を加えています :

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

クラスキャット 人工知能 研究開発支援サービス ⭐️ リニューアルしました 😉

クラスキャット は人工知能に関する各種サービスを提供しています。お気軽にご相談ください :

  • 人工知能導入個別相談会(無償)実施中! [詳細]

  • 人工知能研究開発支援 [詳細]
    1. 自社特有情報を含むチャットボット構築支援
    2. 画像認識 (医療系含む) / 画像生成

  • PoC(概念実証)を失敗させないための支援 [詳細]

お問合せ : 下記までお願いします。

  • クラスキャット セールス・インフォメーション
  • sales-info@classcat.com
  • ClassCatJP

 

 

Model Context Protocol (MCP) : クイックスタート : クライアント開発者向け (Python SDK)

すべての MCP サーバと統合可能な独自クライアントの構築を始めましょう。

このチュートリアルでは、MCP サーバに接続する、LLP-powered チャットボット・クライアントを構築する方法を学習します。

このチュートリアル用の完全なコードは ここ にあります。

 

システム要件

始める前に、システムが以下の要件を満たしているか確認してください :

  • Mac or Windows コンピュータ
  • 最新版 Python がインストールされている
  • 最新版 uv がインストールされている

 

環境のセットアップ

まず、uv を使用して新しい Python プロジェクトを作成します :

# Create project directory
uv init mcp-client
cd mcp-client

# Create virtual environment
uv venv

# Activate virtual environment
# On Windows:
.venv\Scripts\activate
# On Unix or MacOS:
source .venv/bin/activate

# Install required packages
uv add mcp anthropic python-dotenv

# Remove boilerplate files
# On Windows:
del main.py
# On Unix or MacOS:
rm main.py

# Create our main file
touch client.py

 

API キーのセットアップ

Anthropic コンソールからの Anthropic API キーが必要になります。

.env ファイルを作成してそれを保存します :

# Create .env file
touch .env

キーを .env ファイルに追加します :

ANTHROPIC_API_KEY=<your key here>

.env を .gitignore に追加します :

echo ".env" >> .gitignore

 

クライアントの作成

基本的なクライアント構造

まず、インポートをセットアップして基本的なクライアントのクラスを作成しましょう :

import asyncio
from typing import Optional
from contextlib import AsyncExitStack

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

from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()  # load environment variables from .env

class MCPClient:
    def __init__(self):
        # Initialize session and client objects
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self.anthropic = Anthropic()
    # methods will go here

 

サーバ接続管理

次に、MCP サーバに接続するメソッドを実装します :

async def connect_to_server(self, server_script_path: str):
    """Connect to an MCP server

    Args:
        server_script_path: Path to the server script (.py or .js)
    """
    is_python = server_script_path.endswith('.py')
    is_js = server_script_path.endswith('.js')
    if not (is_python or is_js):
        raise ValueError("Server script must be a .py or .js file")

    command = "python" if is_python else "node"
    server_params = StdioServerParameters(
        command=command,
        args=[server_script_path],
        env=None
    )

    stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
    self.stdio, self.write = stdio_transport
    self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))

    await self.session.initialize()

    # List available tools
    response = await self.session.list_tools()
    tools = response.tools
    print("\nConnected to server with tools:", [tool.name for tool in tools])

 

クエリー処理ロジック

そしてクエリー処理とツール呼び出しの処理のためのコア機能を追加しましょう :

async def process_query(self, query: str) -> str:
    """Process a query using Claude and available tools"""
    messages = [
        {
            "role": "user",
            "content": query
        }
    ]

    response = await self.session.list_tools()
    available_tools = [{
        "name": tool.name,
        "description": tool.description,
        "input_schema": tool.inputSchema
    } for tool in response.tools]

    # Initial Claude API call
    response = self.anthropic.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1000,
        messages=messages,
        tools=available_tools
    )

    # Process response and handle tool calls
    final_text = []

    assistant_message_content = []
    for content in response.content:
        if content.type == 'text':
            final_text.append(content.text)
            assistant_message_content.append(content)
        elif content.type == 'tool_use':
            tool_name = content.name
            tool_args = content.input

            # Execute tool call
            result = await self.session.call_tool(tool_name, tool_args)
            final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")

            assistant_message_content.append(content)
            messages.append({
                "role": "assistant",
                "content": assistant_message_content
            })
            messages.append({
                "role": "user",
                "content": [
                    {
                        "type": "tool_result",
                        "tool_use_id": content.id,
                        "content": result.content
                    }
                ]
            })

            # Get next response from Claude
            response = self.anthropic.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1000,
                messages=messages,
                tools=available_tools
            )

            final_text.append(response.content[0].text)

    return "\n".join(final_text)

 

インタラクティブ・チャット・インターフェイス

そしてチャットループとクリーンアップ機能を追加します :

async def chat_loop(self):
    """Run an interactive chat loop"""
    print("\nMCP Client Started!")
    print("Type your queries or 'quit' to exit.")

    while True:
        try:
            query = input("\nQuery: ").strip()

            if query.lower() == 'quit':
                break

            response = await self.process_query(query)
            print("\n" + response)

        except Exception as e:
            print(f"\nError: {str(e)}")

async def cleanup(self):
    """Clean up resources"""
    await self.exit_stack.aclose()

 

メイン・エントリポイント

最後に、メイン実行ロジックを追加します :

async def main():
    if len(sys.argv) < 2:
        print("Usage: python client.py ")
        sys.exit(1)

    client = MCPClient()
    try:
        await client.connect_to_server(sys.argv[1])
        await client.chat_loop()
    finally:
        await client.cleanup()

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

You can find the complete client.py file here.

 

主要なコンポーネントの説明

1. クライアント初期化

  • MCPClient クラスはセッション管理と API クライアントとともに初期化されます。
  • 適切なリソース管理のために AsyncExitStack を使用します。
  • Claude インタラクションのために Anthropic クライアントを構成します。

 

2. サーバ接続

  • Python と Node.js サーバの両方をサポート
  • サーバスクリプトのタイプの検証
  • 適切な通信チャネルのセットアップ
  • セッションを初期化し、利用可能なツールをリストアップ

 

3. クエリー処理

  • 会話コンテキストの保持
  • Claude のレスポンスとツール呼び出しの処理
  • Claude とツールの間のメッセージフローの管理
  • 結果を統合して一貫性のあるレスポンスにする

 

4. 対話型インターフェイス

  • 単純なコマンドライン・インターフェイスの提供
  • ユーザ入力を処理してレスポンスを表示する
  • 基本的なエラー処理を含める
  • 安全な終了 (graceful exit) を可能にする

 

5. リソース管理

  • リソースの適切なクリーンアップ
  • 接続の問題に対するエラー処理
  • 安全な終了 (graceful shutdown) 手続き

 

クライアントの実行

任意の MCP サーバとともにクライアントを実行するには :

uv run client.py path/to/server.py # python server
uv run client.py path/to/build/index.js # node server

クライアントは以下を行います :

  1. 特定のサーバに接続する

  2. 利用可能なツールをリストアップする

  3. 対話的なチャットセッションを開始します、そこでは以下が可能です :
    • クエリーの入力
    • ツール実行の確認
    • Claude からレスポンスを取得する

サーバのクイックスタートからの天気予報サーバに接続した場合、それがどのようなものかの例が以下です :

 

How It Works

クエリーを submit すると :

  1. クライアントはサーバから利用可能なツールのリストを取得します

  2. クエリーはツール説明とともに Claude に送信されます

  3. Claude は (もしあれば) どのツールを使用するか決定します

  4. クライアントはサーバを通して要求されたツール呼び出しを実行します

  5. 結果は Claude に送り返されます

  6. Claude は自然言語のレスポンスを提供します

  7. レスポンスが表示されます

 

以上