MCP入門: Pythonコード例で学ぶAI連携

pip install boto3 python-dotenv mcp fastmcp pydantic anyio

AWS_REGION=ap-northeast-1  # Bedrockが利用可能なリージョン
AWS_ACCESS_KEY_ID=アクセスキー
AWS_SECRET_ACCESS_KEY=シークレットキー
AWS_SESSION_TOKEN=セッショントークン

# simple_mcp.py
from fastmcp import FastMCP
import logging
import sys

# ロガーの設定
logger = logging.getLogger(__name__)
handler = logging.StreamHandler(sys.stderr)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# FastMCPサーバーの作成
mcp = FastMCP("Simple MCP Demo")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers together"""
    logger.info(f"add() tool called: a={a}, b={b}")
    result = a + b
    logger.info(f"add() result: {result}")
    return result

@mcp.tool()
def multiply(a: float, b: float) -> float:
    """Multiply two numbers"""
    logger.info(f"multiply() tool called: a={a}, b={b}")
    result = a * b
    logger.info(f"multiply() result: {result}")
    return result

@mcp.resource("server://status")
def get_server_status() -> str:
    """Get the current server status"""
    logger.info("get_server_status() resource called")
    status = {
        "status": "running",
        "version": "1.0.0",
        "uptime": "N/A"
    }
    logger.info(f"get_server_status() result: {status}")
    return str(status)

if __name__ == "__main__":
    # 標準入出力モードで実行
    logger.info("MCPサーバーを標準入出力（stdio）モードで起動します")
    mcp.run(transport="stdio")

   async def call_bedrock_claude(message: str, tools_info: str, conversation_history: Optional[List[str]] = None) -> str:
       """Call the Claude model through AWS Bedrock"""
       # AWS Bedrockクライアントの初期化
       bedrock = boto3.client(
           service_name='bedrock-runtime',
           region_name=os.getenv('AWS_REGION'),  # .envから環境変数を読み込み
           aws_access_key_id=os.getenv('AWS_ACCESS_KEY_ID'),
           aws_secret_access_key=os.getenv('AWS_SECRET_ACCESS_KEY'),
           aws_session_token=os.getenv('AWS_SESSION_TOKEN')
       )
       
       # 会話履歴の処理
       if conversation_history is None:
           conversation_history = []
       conversation_text = "\n".join(conversation_history) if conversation_history else ""
       
       # プロンプトの生成 - ツール情報を含める
       prompt = f"\n\nHuman: {tools_info}\n\n"
       if conversation_text:
           prompt += f"Previous conversation:\n{conversation_text}\n\n"
       
       # ユーザーのメッセージとツール使用の指示を追加
       prompt += f"User message: {message}\n\n"
       prompt += "If the user's request involves simple math like addition or multiplication, "
       prompt += "you should use the available tools to perform the calculation accurately.\n"
       
       # ツール使用のためのタグ形式を指定
       prompt += "For addition with two numbers, use: <add>number1, number2</add>\n"
       prompt += "For multiplication with two numbers, use: <multiply>number1, number2</multiply>\n"
       prompt += "For requesting a greeting for a specific name, use: <greeting>name</greeting>\n\n"
       prompt += "Assistant:"
       
       # Claude V2用のリクエスト構造
       request_body = {
           "prompt": prompt,
           "max_tokens_to_sample": 1000,
           "temperature": 0.7,
           "anthropic_version": "bedrock-2023-05-31"
       }
       
       # Bedrockを呼び出してレスポンスを取得
       response = bedrock.invoke_model(
           modelId=MODEL_ID,  # "anthropic.claude-v2:1"
           body=json.dumps(request_body).encode('utf-8'),
           contentType="application/json",
           accept="application/json"
       )
       
       # レスポンスを処理して返却
       response_text = response['body'].read().decode('utf-8')
       response_body = json.loads(response_text)
       completion = response_body.get("completion", "No response content")
       logger.info(f"Received response from model: {completion[:30]}...")
       
       return completion

   async def handle_tool_calls(session: ClientSession, claude_response: str) -> str:
       """Handle any tool calls in Claude's response and update the response"""
       updated_response = claude_response
       
       # 足し算ツールの呼び出しを検出 (<add>5, 3</add>のようなパターン)
       add_match = re.search(ADD_PATTERN, claude_response, re.IGNORECASE | re.DOTALL)
       if add_match:
           # 数値を抽出
           a = int(add_match.group(1))
           b = int(add_match.group(2))
           logger.info(f"Detected add tool call: {a} + {b}")
           
           # MCPサーバーの「add」ツールを実行
           result = await session.call_tool("add", arguments={"a": a, "b": b})
           
           # 結果をPydanticモデルから辞書に変換してJSON化
           result_dict = model_to_dict(result)
           result_json = json.dumps(result_dict, cls=MCPJSONEncoder)
           result_obj = json.loads(result_json)
           
           # 結果の値を取得
           if isinstance(result_obj, dict) and "result" in result_obj:
               calc_result = result_obj["result"]
           else:
               calc_result = str(result_obj)
           
           # レスポンス内のツール呼び出しを結果で置換
           updated_response = re.sub(
               ADD_PATTERN,  # <add>数値, 数値</add>
               f"{a} + {b} = {calc_result}",  # 「5 + 3 = 8」という形式に変換
               updated_response,
               flags=re.IGNORECASE | re.DOTALL
           )
       
       # 掛け算ツールの呼び出しを検出 (<multiply>数値, 数値</multiply>)
       multiply_match = re.search(MULTIPLY_PATTERN, claude_response, re.IGNORECASE | re.DOTALL)
       if multiply_match:
           # 掛け算の処理も足し算と同様
           # ...
       
       # 挨拶リソースの呼び出しを検出 (<greeting>名前</greeting>)
       greeting_match = re.search(GREETING_PATTERN, claude_response, re.IGNORECASE | re.DOTALL)
       if greeting_match:
           # 挨拶リソースの処理
           # ...
       
       return updated_response

   async def run():
       """Main function to run the MCP client for Bedrock Claude"""
       logger.info("Starting Bedrock Claude MCP client")
       
       # MCPサーバーパラメータの設定 - simple_mcp.pyをサブプロセスとして起動
       server_params = StdioServerParameters(
           command="python",
           args=["simple_mcp.py"],  # MCPサーバースクリプト
           env=None
       )
       
       try:
           # MCPサーバーとstdio経由で接続
           async with stdio_client(server_params) as (read, write):
               logger.info("Connected to simple_mcp.py server")
               
               # MCPクライアントセッションの確立
               async with ClientSession(
                   read, 
                   write, 
                   sampling_callback=sampling_callback  # AIモデル呼び出しコールバック
               ) as session:
                   # セッションの初期化（MCP handshake）
                   logger.info("Initializing MCP session")
                   await session.initialize()
                   
                   # 利用可能なツールとリソースの情報をサーバーから取得
                   logger.info("Listing available tools and resources")
                   tools_info = await create_tools_info(session)
                   logger.info(f"Tools info for Claude:\n{tools_info}")
                   
                   # 対話ループの開始
                   print("\nType your message to Claude (or 'exit' to quit):")
                   print("You can ask Claude to perform calculations or use available tools.")
                   print("For example: 'What is 5 + 3?' or 'Can you multiply 6 and 7?'")
                   
                   # ユーザー入力を受け付けるループ
                   while True:
                       # ユーザー入力の非同期取得
                       user_input = await asyncio.to_thread(input, "> ")
                       if user_input.lower() == 'exit':
                           break
                       
                       logger.info(f"Received user input: {user_input}")
                       
                       # sampling_callbackを通してClaudeにリクエスト送信
                       result = await sampling_callback(
                           context=RequestContext(...),  # 必要なコンテキスト情報
                           params=types.CreateMessageRequestParams(...)  # リクエストパラメータ
                       )
                       
                       # 応答の表示
                       print("\n===== Claude's response =====")
                       if isinstance(result, types.CreateMessageResult):
                           content = result.content
                           if isinstance(content, types.TextContent):
                               print(content.text)  # テキスト応答を表示
                           else:
                               print("Response is not text content")
                       else:
                           print(f"Error: {result.message}")
                       print("===========================\n")
       
       except Exception as e:
           logger.error(f"Error in run function: {str(e)}")
           raise
       
       logger.info("Bedrock Claude MCP client stopped")

python client.py

> 5足す3はいくつですか？

INFO - Received response from model:  5足す3は<add>5, 3</add>です。...
INFO - Detected add tool call: 5 + 3
INFO     Processing request of type CallToolRequest                       server.py:534
INFO - add() tool called: a=5, b=3
INFO     add() tool called: a=5, b=3                                   simple_mcp.py:19
INFO - add() result: 8
INFO     add() result: 8                                               simple_mcp.py:21
===== Claude's response =====
   5 + 3 = 8 です。
===========================