Step-by-Step Guide to Build Your Own MCP Server and MCP Client with UI Using Python

Python を使って独自の MCP サーバーと UI 付き MCP クライアントを構築するステップバイステップガイド

この包括的なガイドでは、Python を使って Model Context Protocol（MCP）サーバーと、グラフィカルユーザーインターフェースを備えたクライアントを作成する手順を説明します。MCP は、言語モデルとツールプロバイダー間の通信を促進し、強力な AI ベースのアプリケーションを実現するために設計されています。このチュートリアルを終えると、サーバーとクライアントの両方のコンポーネントを備え、シームレスに連携できる実用的な MCP システムが完成します。

前提条件と環境のセットアップ

開発を始める前に、お使いのシステムが次の要件を満たしていることを確認してください。

• 最新バージョンの Python がインストールされた Windows または Mac のコンピューター
• Python プログラミングの知識
• Python の非同期プログラミングに関する基本的な理解
• UI 開発の概念に関する知識

まず、必要なツールとパッケージを使って開発環境をセットアップしましょう。

# Install uv tool if not already installed
pip install uv

uv ツールを使うと、Python プロジェクトを効率よく作成および管理できます。これは、このガイド全体を通して使用する最新の Python パッケージマネージャー兼仮想環境ツールです[^1][^2]。

MCP サーバーの構築

ステップ 1：サーバープロジェクトの作成

任意のプロジェクトディレクトリに移動し、新しい MCP サーバープロジェクトを作成します。

cd "your_desired_directory"
uvx create-mcp-server

作成中に、名前、説明、設定オプションなどのプロジェクトの詳細を求められます。これにより、必要なファイルと依存関係をすべて含むプロジェクト構成が生成されます[^1]。

ステップ 2：仮想環境のセットアップ

プロジェクトを作成したら、プロジェクトディレクトリに移動して仮想環境をセットアップします。

cd server-project-name
uv sync --dev --all-extras

このコマンドにより、すべての依存関係が仮想環境に正しくインストールされます[^1]。

ステップ 3：プロジェクト構成の確認

生成されたサーバープロジェクトは、次のような構成になっています。

SERVER-PROJECT-NAME/
├── .venv/
├── Lib/
│ └── site-packages/
├── Scripts/
├── src/
│ └── server_project_name/
│ ├── __pycache__/
│ ├── __init__.py
│ └── server.py
├── .gitignore
├── pyproject.toml
├── README.md
└── uv.lock

注目すべき主要なファイルは、src/server_project_name/ ディレクトリ内にあります。特に、サーバーの実装を含む server.py が重要です[^1]。

ステップ 4：サーバー機能の実装

server.py ファイルを開き、サーバー機能を実装します。基本的な MCP サーバーでは、クライアントが呼び出せるツールを定義します。以下は、加算ツールを実装する簡単な例です。

from mcp import Server, ToolCall, ToolResult

class MyMCPServer(Server):
    async def handle_tool_call(self, tool_call: ToolCall) -> ToolResult:
        if tool_call.name == "add":
            # Extract arguments
            a = tool_call.arguments.get("a")
            b = tool_call.arguments.get("b")
            
            # Perform addition
            result = a + b
            
            # Return the result
            return ToolResult(content=str(result))
        
        # Handle unknown tool
        return ToolResult(error=f"Unknown tool: {tool_call.name}")
    
    async def list_tools(self):
        # Define available tools
        tools = [
            {
                "name": "add",
                "description": "Adds two numbers together",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "a": {"type": "number"},
                        "b": {"type": "number"}
                    },
                    "required": ["a", "b"]
                }
            }
        ]
        return tools

# Run the server
if __name__ == "__main__":
    server = MyMCPServer()
    server.run()

この例では、クライアントが呼び出せる簡単な加算ツールを備えたサーバーを作成します[^3]。

ステップ 5：サーバーの実行

MCP サーバーを実行するには、次のようにします。

python -m server_project_name.server

デフォルトでは、サーバーは localhost のポート 8000 で実行されます。必要に応じて、サーバー設定で変更できます[^3]。

MCP クライアントの構築

ステップ 1：クライアントプロジェクトの作成

MCP クライアント用の新しいディレクトリを作成します。

uv init mcp-client
cd mcp-client

仮想環境をセットアップし、必要なパッケージをインストールします。

uv venv
# Activate the virtual environment
# On Windows:
.venv\Scripts\activate
# On macOS/Linux:
source .venv/bin/activate

# Install required packages
uv add mcp anthropic python-dotenv

クライアントの実装用に新しいファイルを作成します。

touch client.py

ステップ 2：クライアントの実装

サーバーに接続してそのツールを呼び出せる、基本的な MCP クライアントを作成します。以下は実装例です。

import asyncio
from typing import Optional
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from dotenv import load_dotenv

load_dotenv()  # Load environment variables

class MCPClient:
    def __init__(self):
        # Initialize session and client objects
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()

    async def connect_to_server(self, server_script_path: str):
        """Connect to an MCP server"""
        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])
        return tools

    async def call_add_tool(self, a: int, b: int):
        """Call the add tool on the server"""
        if not self.session:
            raise RuntimeError("Not connected to server")
            
        result = await self.session.call_tool(
            name="add",
            arguments={"a": a, "b": b}
        )
        return result.content[^0].text

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

async def main():
    client = MCPClient()
    try:
        await client.connect_to_server("path/to/your/server.py")
        result = await client.call_add_tool(4, 5)
        print(f"Result: {result}")
    finally:
        await client.cleanup()

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

このクライアントは MCP サーバーに接続し、引数として 4 と 5 を指定して「add」ツールを呼び出し、結果を出力します[^2][^3]。

Tkinter を使って MCP クライアントに UI を追加する

ステップ 1：基本的な UI のセットアップ

次に、Tkinter を使ってグラフィカルユーザーインターフェースを追加し、クライアントを拡張しましょう。まず、Python のインストールに Tkinter が含まれていない場合は、インストールします。

uv add tk

次に、client_ui.py という新しいファイルを作成します。

import asyncio
import tkinter as tk
from tkinter import ttk, messagebox
from client import MCPClient  # Import the client class we created earlier

class MCPClientUI:
    def __init__(self, root):
        self.root = root
        self.root.title("MCP Client")
        self.root.geometry("500x400")
        
        self.client = MCPClient()
        self.connected = False
        
        self.setup_ui()
        
    def setup_ui(self):
        # Create main frame
        main_frame = ttk.Frame(self.root, padding="10")
        main_frame.pack(fill=tk.BOTH, expand=True)
        
        # Server connection section
        server_frame = ttk.LabelFrame(main_frame, text="Server Connection", padding="10")
        server_frame.pack(fill=tk.X, pady=5)
        
        ttk.Label(server_frame, text="Server Script Path:").grid(row=0, column=0, sticky=tk.W, pady=5)
        self.server_path = tk.StringVar(value="path/to/your/server.py")
        ttk.Entry(server_frame, textvariable=self.server_path, width=40).grid(row=0, column=1, padx=5, pady=5)
        
        self.connect_btn = ttk.Button(server_frame, text="Connect", command=self.connect_to_server)
        self.connect_btn.grid(row=0, column=2, padx=5, pady=5)
        
        # Tool section
        tool_frame = ttk.LabelFrame(main_frame, text="Add Tool", padding="10")
        tool_frame.pack(fill=tk.X, pady=5)
        
        ttk.Label(tool_frame, text="Number A:").grid(row=0, column=0, sticky=tk.W, pady=5)
        self.num_a = tk.IntVar(value=0)
        ttk.Entry(tool_frame, textvariable=self.num_a, width=10).grid(row=0, column=1, padx=5, pady=5)
        
        ttk.Label(tool_frame, text="Number B:").grid(row=0, column=2, sticky=tk.W, pady=5)
        self.num_b = tk.IntVar(value=0)
        ttk.Entry(tool_frame, textvariable=self.num_b, width=10).grid(row=0, column=3, padx=5, pady=5)
        
        self.calculate_btn = ttk.Button(tool_frame, text="Calculate", command=self.call_add_tool, state=tk.DISABLED)
        self.calculate_btn.grid(row=0, column=4, padx=5, pady=5)
        
        # Results section
        result_frame = ttk.LabelFrame(main_frame, text="Results", padding="10")
        result_frame.pack(fill=tk.BOTH, expand=True, pady=5)
        
        self.result_text = tk.Text(result_frame, height=10, width=50)
        self.result_text.pack(fill=tk.BOTH, expand=True)
        
    def connect_to_server(self):
        server_path = self.server_path.get()
        self.result_text.insert(tk.END, f"Connecting to server at {server_path}...\n")
        
        # Create and run a coroutine to connect to the server
        async def connect():
            try:
                tools = await self.client.connect_to_server(server_path)
                self.connected = True
                self.root.after(0, self.update_ui_after_connect, tools)
            except Exception as e:
                self.root.after(0, lambda: self.show_error(f"Connection error: {str(e)}"))
        
        asyncio.run(connect())
    
    def update_ui_after_connect(self, tools):
        self.result_text.insert(tk.END, f"Connected to server. Available tools: {[tool.name for tool in tools]}\n")
        self.connect_btn.config(text="Connected", state=tk.DISABLED)
        self.calculate_btn.config(state=tk.NORMAL)
    
    def call_add_tool(self):
        if not self.connected:
            self.show_error("Not connected to server")
            return
            
        try:
            a = self.num_a.get()
            b = self.num_b.get()
            
            self.result_text.insert(tk.END, f"Calculating {a} + {b}...\n")
            
            # Create and run a coroutine to call the add tool
            async def add():
                try:
                    result = await self.client.call_add_tool(a, b)
                    self.root.after(0, lambda: self.show_result(result))
                except Exception as e:
                    self.root.after(0, lambda: self.show_error(f"Tool call error: {str(e)}"))
            
            asyncio.run(add())
        except Exception as e:
            self.show_error(f"Input error: {str(e)}")
    
    def show_result(self, result):
        self.result_text.insert(tk.END, f"Result: {result}\n")
    
    def show_error(self, message):
        messagebox.showerror("Error", message)
        self.result_text.insert(tk.END, f"Error: {message}\n")
    
    def cleanup(self):
        asyncio.run(self.client.cleanup())
        self.root.destroy()

if __name__ == "__main__":
    root = tk.Tk()
    app = MCPClientUI(root)
    root.protocol("WM_DELETE_WINDOW", app.cleanup)
    root.mainloop()

この UI では、サーバーのパスと加算処理の値を入力するフィールド、サーバーへの接続と結果の計算を行うボタン、出力を表示するテキスト領域を利用できます[^4]。

ステップ 2：Tkinter での非同期処理

Tkinter は、MCP クライアントが依存している asyncio と連携するようには設計されていません。上記の解決策では、asyncio の処理をブロッキングモードで実行するシンプルな方法を使っています。この方法は単純な処理には適していますが、より複雑なアプリケーションには理想的ではありません。

より堅牢な解決策として、非同期処理をより適切に扱える拡張版を作成しましょう。

import asyncio
import tkinter as tk
from tkinter import ttk, messagebox
from client import MCPClient
import threading

class AsyncTkApp:
    def __init__(self, root):
        self.root = root
        self.root.title("Advanced MCP Client")
        self.root.geometry("550x450")
        
        self.client = MCPClient()
        self.connected = False
        self.loop = asyncio.new_event_loop()
        
        # Start a new thread for running the async event loop
        self.thread = threading.Thread(target=self._start_loop, daemon=True)
        self.thread.start()
        
        self.setup_ui()
    
    def _start_loop(self):
        asyncio.set_event_loop(self.loop)
        self.loop.run_forever()
    
    def setup_ui(self):
        # Create main frame with padding
        main_frame = ttk.Frame(self.root, padding="15")
        main_frame.pack(fill=tk.BOTH, expand=True)
        
        # Server connection section
        server_frame = ttk.LabelFrame(main_frame, text="Server Connection", padding="10")
        server_frame.pack(fill=tk.X, pady=8)
        
        ttk.Label(server_frame, text="Server Script Path:").grid(row=0, column=0, sticky=tk.W, pady=5)
        self.server_path = tk.StringVar(value="path/to/your/server.py")
        ttk.Entry(server_frame, textvariable=self.server_path, width=40).grid(row=0, column=1, padx=5, pady=5)
        
        self.connect_btn = ttk.Button(server_frame, text="Connect", command=self.connect_to_server)
        self.connect_btn.grid(row=0, column=2, padx=5, pady=5)
        
        # Tool section with improved layout
        tool_frame = ttk.LabelFrame(main_frame, text="Add Tool", padding="10")
        tool_frame.pack(fill=tk.X, pady=8)
        
        ttk.Label(tool_frame, text="Number A:").grid(row=0, column=0, sticky=tk.W, pady=5)
        self.num_a = tk.IntVar(value=0)
        ttk.Entry(tool_frame, textvariable=self.num_a, width=10).grid(row=0, column=1, padx=5, pady=5)
        
        ttk.Label(tool_frame, text="Number B:").grid(row=0, column=2, sticky=tk.W, pady=5)
        self.num_b = tk.IntVar(value=0)
        ttk.Entry(tool_frame, textvariable=self.num_b, width=10).grid(row=0, column=3, padx=5, pady=5)
        
        self.calculate_btn = ttk.Button(tool_frame, text="Calculate", command=self.call_add_tool, state=tk.DISABLED)
        self.calculate_btn.grid(row=0, column=4, padx=5, pady=5)
        
        # Results section with scrollbar
        result_frame = ttk.LabelFrame(main_frame, text="Results", padding="10")
        result_frame.pack(fill=tk.BOTH, expand=True, pady=8)
        
        # Add a scrollbar to the text widget
        scrollbar = ttk.Scrollbar(result_frame)
        scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
        
        self.result_text = tk.Text(result_frame, height=12, width=50, yscrollcommand=scrollbar.set)
        self.result_text.pack(fill=tk.BOTH, expand=True)
        scrollbar.config(command=self.result_text.yview)
        
        # Status bar
        status_frame = ttk.Frame(main_frame)
        status_frame.pack(fill=tk.X, pady=5)
        self.status_var = tk.StringVar(value="Ready")
        status_label = ttk.Label(status_frame, textvariable=self.status_var, anchor=tk.W)
        status_label.pack(side=tk.LEFT)
    
    def connect_to_server(self):
        server_path = self.server_path.get()
        self.log_message(f"Connecting to server at {server_path}...")
        self.status_var.set("Connecting...")
        
        async def connect_async():
            try:
                tools = await self.client.connect_to_server(server_path)
                self.root.after(0, lambda: self.update_ui_after_connect(tools))
            except Exception as e:
                self.root.after(0, lambda: self.show_error(f"Connection error: {str(e)}"))
        
        # Schedule the coroutine to run in the asyncio event loop
        asyncio.run_coroutine_threadsafe(connect_async(), self.loop)
    
    def update_ui_after_connect(self, tools):
        self.connected = True
        self.log_message(f"Connected to server. Available tools: {[tool.name for tool in tools]}")
        self.connect_btn.config(text="Connected", state=tk.DISABLED)
        self.calculate_btn.config(state=tk.NORMAL)
        self.status_var.set("Connected")
    
    def call_add_tool(self):
        if not self.connected:
            self.show_error("Not connected to server")
            return
            
        try:
            a = self.num_a.get()
            b = self.num_b.get()
            
            self.log_message(f"Calculating {a} + {b}...")
            self.status_var.set("Calculating...")
            
            async def add_async():
                try:
                    result = await self.client.call_add_tool(a, b)
                    self.root.after(0, lambda: self.show_result(result))
                except Exception as e:
                    self.root.after(0, lambda: self.show_error(f"Tool call error: {str(e)}"))
            
            # Schedule the coroutine to run in the asyncio event loop
            asyncio.run_coroutine_threadsafe(add_async(), self.loop)
        except Exception as e:
            self.show_error(f"Input error: {str(e)}")
    
    def log_message(self, message):
        self.result_text.insert(tk.END, f"{message}\n")
        self.result_text.see(tk.END)  # Scroll to the end
    
    def show_result(self, result):
        self.log_message(f"Result: {result}")
        self.status_var.set("Ready")
    
    def show_error(self, message):
        messagebox.showerror("Error", message)
        self.log_message(f"Error: {message}")
        self.status_var.set("Error occurred")
    
    def cleanup(self):
        async def cleanup_async():
            await self.client.cleanup()
            self.loop.stop()
        
        # Schedule cleanup in the event loop
        asyncio.run_coroutine_threadsafe(cleanup_async(), self.loop)
        self.root.after(100, self.root.destroy)  # Give time for cleanup to complete

if __name__ == "__main__":
    root = tk.Tk()
    app = AsyncTkApp(root)
    root.protocol("WM_DELETE_WINDOW", app.cleanup)
    root.mainloop()

この拡張版では、asyncio のイベントループに別のスレッドを使用するため、UI をブロックせずに非同期処理をより適切に扱えます[^4]。

システム全体のテスト

MCP サーバーとクライアントをテストするには、次の手順を実行します。

1. 1 つ目のターミナルで MCP サーバーを起動します。

cd server-project-directory
python -m server_project_name.server

2. 別のターミナルで MCP クライアント UI を実行します。

cd mcp-client
python client_ui.py

3. クライアント UI で、次の操作を行います。
   • サーバースクリプトへのパスを入力します
   • 「Connect」をクリックして接続を確立します
   • 入力欄に 2 つの数値を入力します
   • 「Calculate」をクリックして、サーバーにリクエストを送信します
   • 結果表示用のテキスト領域で結果を確認します

システムの拡張

基本的なサーバーとクライアントが動作したら、次の方法でシステムを拡張できます。

1. テキスト処理やデータ分析関数など、サーバーにツールを追加する
2. 機能を追加し、エラー処理を改善してクライアント UI を強化する
3. 認証とセキュリティ対策を実装する
4. 複数の同時接続に対応する
5. グラフ、チャートなどの視覚化を備えた、より高度な UI を作成する

まとめ

このステップごとのガイドでは、グラフィカルユーザーインターフェースを備えた MCP サーバーとクライアントのシステムを構築しました。Model Context Protocol によってクライアントとサーバー間の強力な通信が可能になり、AI やその他のツールを活用する幅広いアプリケーションを実現できます。

構築したシステムでは、ツールの定義、クライアントからの呼び出し、ユーザーフレンドリーなインターフェースでの結果表示という、MCP の基本概念を実演しています。この知識は、さまざまなユースケースに対応する、より複雑な MCP ベースのアプリケーションを開発するための確かな基盤となります。

MCP は進化を続けているため、プロトコルの最新動向を把握しておくことで、さらに強力で効率的なアプリケーションを構築できます。ここで学んだ基本概念は、技術が進歩しても引き続き応用できます。