Building a Watsonx.ai Chatbot Server with MCP in Python

Python で MCP を使って Watsonx.ai チャットボットサーバーを構築する

この詳細なチュートリアルでは、IBM Watsonx.ai を活用し、Model Context Protocol（MCP）Python SDK 経由で公開する、本番環境に対応したプロフェッショナルなチャットボットサーバーの作成方法を学びます。最後まで進めると、MCP 対応クライアント（Claude Desktop やカスタム Python クライアントなど）から標準化された「チャット」ツールとして呼び出せる、再利用可能な MCP サービスが完成します。

以下の手順を説明します。

• 環境を順を追ってセットアップする
• 依存関係をインストールし、認証情報を安全に管理する
• 読みやすく、十分に文書化された Python コードを書く
• Watsonx.ai の推論を MCP ツールとして公開する
• サーバーを起動、テストし、問題を解決する
• サービスを拡張し、堅牢化するためのヒント

はじめに

IBM Watsonx.ai は IBM Cloud を通じて最先端の大規模言語モデル（LLM）推論を提供し、Model Context Protocol（MCP）はアプリケーションがツール、リソース、プロンプトを LLM クライアントに公開する方法を標準化します。この 2 つを組み合わせることで、次のメリットが得られます。

• モジュール性： チャットボットのロジックをクライアントの実装から切り離せます。
• 再利用性： どの MCP 対応クライアントからも、同じ「チャット」エンドポイントを呼び出せます。
• 迅速な反復： ライブリロードに対応した開発用インスペクターが組み込まれています。

社内ヘルプデスク用ボットを構築する場合でも、公開チャットボット API を構築する場合でも、このパターンは簡単に拡張、適応できます。

前提条件

始める前に、以下が揃っていることを確認してください。

• IBM Cloud Watsonx.ai の認証情報： API キー、サービス URL、プロジェクト ID
• Python 3.9+（パフォーマンスと型付けの改善のため、3.11+ を推奨します）
• pip（Python パッケージインストーラー）
• 仮想環境ツール（venv または virtualenv）
• コマンドラインの基本的な知識（Linux/macOS/Windows WSL）

以下の Python パッケージをインストールします。

• python-dotenv – .env ファイルから環境変数を読み込みます
• ibm-watsonx-ai – IBM 公式の Watsonx.ai SDK
• mcp[cli] – MCP Python SDK と CLI ツール

プロジェクト構成

プロジェクト用の新しいディレクトリを作成します。最終的なディレクトリ構成は次のようになります。

watsonx-mcp-server/
├── .env
├── .gitignore
├── requirements.txt
└── server.py

• .env — 秘密の認証情報（ソース管理に絶対にコミットしないでください！）
• .gitignore — .env、__pycache__、.venv を除外します
• requirements.txt — バージョンを固定した依存関係の一覧
• server.py — MCP チャットボットサーバーの完全な実装

環境のセットアップ

1. 仮想環境を作成して有効化する

   python3 -m venv .venv
   source .venv/bin/activate      # macOS/Linux
   .venv\Scripts\activate.bat     # Windows
   

2. 依存関係のバージョンを固定してインストールする

requirements.txt の場合

python-dotenv>=0.21.0
ibm-watsonx-ai==1.3.8
mcp[cli]>=1.6.0

pip install --upgrade pip
pip install -r requirements.txt

3. 認証情報を保護する
   • プロジェクトのルートに .env という名前のファイルを作成します。
   • Watsonx.ai の詳細情報を追加します。

     WATSONX_APIKEY=your-ibm-watsonx-api-key
     WATSONX_URL=https://api.your-region.watsonx.ai
     PROJECT_ID=your-watsonx-project-id
     

   • .env（および .venv/、__pycache__/）を .gitignore に追加します。

     .env
     .venv/
     __pycache__/
     

---

チャットボットサーバー（server.py）を書く

server.py を開き、以下のセクションに沿って進めてください。

インポートと設定

# server.py

import os
import logging
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP

# IBM Watsonx.ai SDK
from ibm_watsonx_ai import APIClient, Credentials
from ibm_watsonx_ai.foundation_models import ModelInference
from ibm_watsonx_ai.metanames import GenTextParamsMetaNames as GenParams

• デバッグを容易にするため、ログを設定します。
• 環境変数を早い段階で読み込みます。

# Load .env variables
load_dotenv()

# Fetch credentials
API_KEY    = os.getenv("WATSONX_APIKEY")
URL        = os.getenv("WATSONX_URL")
PROJECT_ID = os.getenv("PROJECT_ID")
MODEL_ID   = os.getenv("MODEL_ID", "ibm/granite-13b-instruct-v2")

認証情報の検証とクライアントの初期化

# Validate env vars
for name, val in [
    ("WATSONX_APIKEY", API_KEY),
    ("WATSONX_URL", URL),
    ("PROJECT_ID", PROJECT_ID)
]:
    if not val:
        raise RuntimeError(f"{name} is not set. Please add it to your .env file.")

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)

# Initialize IBM credentials & client
creds  = Credentials(url=URL, api_key=API_KEY)
client = APIClient(credentials=creds, project_id=PROJECT_ID)

# Initialize the inference model
model = ModelInference(
    model_id=MODEL_ID,
    credentials=creds,
    project_id=PROJECT_ID
)

logging.info(
    f"Initialized Watsonx.ai model '{MODEL_ID}' "
    f"for project '{PROJECT_ID}'."
)

MCP の「チャット」ツールを定義する

# Create the MCP server instance
mcp = FastMCP("Watsonx Chatbot Server")

@mcp.tool()
def chat(query: str) -> str:
    """
    MCP tool: generate a chatbot response via Watsonx.ai

    :param query: User's input message
    :return: Watsonx.ai generated response
    """
    logging.info("Received chat query: %r", query)

    # Define generation parameters
    params = {
        GenParams.DECODING_METHOD: "greedy",
        GenParams.MAX_NEW_TOKENS:   200,
    }

    # Run the model
    try:
        # Request the full JSON response rather than just a string
        resp = model.generate_text(
            prompt=query,
            params=params,
            raw_response=True
        )
        print("AI raw response:", resp)

        # Extract the generated text from the dict
        text = resp["results"][0]["generated_text"].strip()
        logging.info("Generated response: %r", text)
        return text

    except Exception as e:
        logging.error("Inference error: %s", e, exc_info=True)
        return f"Error generating response: {e}"

---

プロンプトとリソース

リソースを公開する
リソースを使うと、動的データを LLM のコンテキストに注入できます。たとえば、ここではパーソナライズされた挨拶を公開します。

# Expose a greeting resource that dynamically constructs a personalized greeting.
@mcp.resource("greeting://patient/{name}")
def get_greeting(name: str) -> str:
    """
    Return a medical‑style greeting for the given patient name.
    
    :param name: The patient's name.
    :return: A personalized greeting.
    """
    return f"Hello {name}, I’m your medical assistant. How can I help you today?"

プロンプトを追加する（任意）
プロンプトを使うと、再利用可能なテンプレートを定義できます。ここでは、モデルに症状の評価を依頼する簡単な医療チャットの例を示します。

from mcp.server.fastmcp.prompts import base

@mcp.prompt()
def assess_symptoms(symptoms: str) -> str:
    """
    Prompt template for symptom assessment.
    
    :param symptoms: Description of patient symptoms.
    :return: A prompt asking the LLM to analyze and suggest next steps.
    """
    return (
        f"{base}\n"
        "You are a qualified medical assistant. The patient reports the following symptoms:\n"
        f"{symptoms}\n\n"
        "Please provide possible causes, recommended next steps, and when to seek immediate care."
    )

---

エントリーポイントとログ

if __name__ == "__main__":
    # Start the MCP server (blocking call)
    logging.info("Starting MCP server on STDIO transport...")
    mcp.run()

これで完了です！以下が揃いました。

1. Watsonx.ai の raw_response を正しく解析する、動作する chat ツール。
2. パーソナライズされた挨拶を注入するための リソース。
3. 症状の評価に使う、任意の プロンプト テンプレート。

サーバーの起動とテスト

まず、環境を読み込みます。

source .venv/bin/activate

mcp dev による開発モード

最も手早く反復開発する方法は次のとおりです。

mcp dev server.py

• コードの変更時にライブリロード
• http://localhost:6274/ のインスペクター UI
• chat ツールの対話型テスト

サーバーとの対話方法

次の URL にアクセスします。 http://localhost:6274/

ブラウザーでインスペクターを開くと、サーバーのテストを容易にするためのいくつかの重要なセクションが表示されます。

MCP Inspector インターフェースの上部に移動すると、次のように表示されています。

Transport Type: STDIO  
Command: python  
Arguments: run --with mcp mcp run server.py

server.py は通常の pip ベースの仮想環境を使用するスタンドアロンのスクリプトであるため、MCP Inspector の設定を次のように修正する必要があります。

Transport Type: STDIO  
Command: python  
Arguments: server.py

次に「Connect」をクリックすると、Python インタープリターを使用してサーバーが正しく起動します。

次に「tools」から「list tools」に移動し、「chat」をクリックします。クエリに次のように入力できます。

What is watsonx in IBM?

MCP Inspector でプロンプトを使用する

インスペクターのサイドバーで、Prompts → List prompts を展開します。

assess_symptoms が表示されます。

プロンプトペインには、引数を入力できるフォームまたは JSON エディターがあります。

code パラメーターを指定します。例：

persistent dull ache, stiffness, and general back pain

Get Prompt をクリックします。

次の結果が返されます。

{
  "messages": [
    {
      "role": "user",
      "content": {
        "type": "text",
        "text": "<module 'mcp.server.fastmcp.prompts.base' from '/mnt/c/Blog/Watsonx-ai-Chatbot-Server-with-MCP/.venv/lib/python3.11/site-packages/mcp/server/fastmcp/prompts/base.py'>\nYou are a qualified medical assistant. The patient reports the following symptoms:\npersistent dull ache, stiffness, and general back pain\n\nPlease provide possible causes, recommended next steps, and when to seek immediate care."
      }
    }
  ]
}

リソースへのアクセス

上記の設定を使用してインスペクターでサーバーが起動したら、URI を指定して登録済みのリソースを呼び出せます。

1. リソース操作ペインを開く MCP Inspector のサイドバーで「Resources」→「Resource Templates」をクリックします。「List templates」をクリックし、get_greeting を選択します。
2. リソース URI を入力する 入力フィールドに次の名前を入力します。 John
3. リソースを呼び出す 「Read Resource」をクリックします。 インスペクターはその URI を @mcp.resource("greeting://{name}") ハンドラーに送信します。
4. 応答を確認する 次のように表示されます。

{
  "contents": [
    {
      "uri": "greeting://patient/John",
      "mimeType": "text/plain",
      "text": "Hello John, I’m your medical assistant. How can I help you today?"
    }
  ]
}

これにより、動的なあいさつリソースが正しく接続され、要求に応じてパーソナライズされた出力を返すことが確認できます。

直接実行

インスペクターを使わずに実行するには、次のようにします。

python server.py

STDIO で MCP 形式のリクエストを待機し、何も表示せずに動作します。

Python クライアントの例

これを server.py と同じ場所に client.py として保存します。

# client.py
import asyncio
from mcp import ClientSession
from mcp.client.stdio import stdio_client
from mcp import StdioServerParameters

async def main():
    server_params = StdioServerParameters(command="python", args=["server.py"])
    async with stdio_client(server_params) as (reader, writer):
        async with ClientSession(reader, writer) as session:
            await session.initialize()
            # Call the chat tool
            user_msg = "Hello, how are you today?"
            response = await session.call_tool("chat", arguments={"query": user_msg})
            print("Bot:", response)

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

一方のターミナルでサーバーを実行します（python server.py）。

次に、もう一方のターミナルで次を実行します。

python client-tool.py

次のように表示されます。

Bot: meta=None content=[TextContent(type='text', text='I am good. Thank you.', annotations=None)] isError=False  

Flask と MCP を使った Watsonx.ai 医療チャットボットの構築

それでは、次の技術を使って Web ベースの医療チャットボットを作成していきましょう。

• プロンプトとツール呼び出しを構成する MCP（Modular Chat Protocol）
• LLM 推論を行う Watsonx.ai
• フロントエンド UI を提供する Flask

最後まで進めると、次の機能を備えた Flask アプリが完成します。

1. ユーザーを名前で歓迎する
2. 症状を収集する
3. MCP を介して診断プロンプトを生成する
4. Watsonx.ai を呼び出して医療に関するアドバイスを取得する
5. スタイルを適用した Web ページに評価を表示する

プロジェクト構成

├─ chatbot.py
├─ server.py                # Your MCP server implementation
├─ static/
│   └─ assets/
│       └─ watsonx-wallpaper.jpg
└─ templates/
    ├─ base.html
    ├─ home.html
    ├─ symptoms.html
    └─ diagnosis.html

すべてのフロントエンドテンプレートは templates/ に保存します。壁紙などの静的アセットは、Flask の static/ ディレクトリに配置します。

---

ステップ 1：chatbot.py の作成

このスクリプトは、インポート時に長時間維持される MCP クライアントセッションを 1 つ初期化し、リクエスト間で再利用します。

import os
import atexit
import asyncio
from flask import Flask, render_template, request, redirect, url_for, session
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# Flask app setup
app = Flask(__name__)
app.secret_key = os.environ.get("SECRET_KEY", os.urandom(24))

# MCP server parameters
SERVER_PARAMS = StdioServerParameters(command="python", args=["server.py"], env=None)

# Dedicated asyncio loop for MCP
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

# Globals for client and session contexts
_stdio_ctx = None
_session_ctx = None
SESSION = None

async def _init_session():
    global _stdio_ctx, _session_ctx, SESSION
    _stdio_ctx = stdio_client(SERVER_PARAMS)
    _reader, _writer = await _stdio_ctx.__aenter__()
    _session_ctx = ClientSession(_reader, _writer)
    SESSION = await _session_ctx.__aenter__()
    await SESSION.initialize()

# Initialize once at import
loop.run_until_complete(_init_session())
app.logger.info("MCP client session initialized once.")

async def _close_session():
    if _session_ctx:
        await _session_ctx.__aexit__(None, None, None)
    if _stdio_ctx:
        await _stdio_ctx.__aexit__(None, None, None)

atexit.register(lambda: loop.run_until_complete(_close_session()))

# Helper: fetch greeting text
def fetch_greeting(name: str) -> str:
    resp = loop.run_until_complete(SESSION.read_resource(f"greeting://patient/{name}"))
    contents = getattr(resp, 'contents', None)
    if isinstance(contents, list):
        return "\n".join(c.text for c in contents).strip()
    return str(resp)

# Helper: assess symptoms via chat tool
def assess_symptoms(symptoms: str) -> str:
    prompt_resp = loop.run_until_complete(
        SESSION.get_prompt("assess_symptoms", arguments={"symptoms": symptoms})
    )
    # Extract clean text from prompt_resp.messages
    msgs = getattr(prompt_resp, 'messages', None)
    if msgs:
        lines = []
        for m in msgs:
            txt = m.content.text if hasattr(m.content, 'text') else str(m.content)
            if txt.startswith("<module"):
                txt = txt.split("\n", 1)[1]
            lines.append(txt)
        diagnosis_prompt = "\n".join(lines).strip()
    else:
        diagnosis_prompt = str(prompt_resp)

    tool_resp = loop.run_until_complete(
        SESSION.call_tool("chat", arguments={"query": diagnosis_prompt})
    )
    cont = getattr(tool_resp, 'content', None)
    if isinstance(cont, list):
        return "\n".join(c.text for c in cont).strip()
    return str(cont).strip()

# Flask routes
@app.route("/", methods=["GET", "POST"])
def home():
    if request.method == "POST":
        session['name'] = request.form['name']
        return redirect(url_for('symptoms'))
    return render_template("home.html")

@app.route("/symptoms", methods=["GET", "POST"])
def symptoms():
    name = session.get('name')
    if not name:
        return redirect(url_for('home'))

    if request.method == "POST":
        diag = assess_symptoms(request.form['symptoms'])
        return render_template("diagnosis.html", diagnosis=diag)

    greet = fetch_greeting(name)
    return render_template("symptoms.html", greeting=greet)

if __name__ == "__main__":
    app.run(debug=True)

説明：

• HTTP リクエストごとに再接続しないよう、インポート時に stdio_client と ClientSession を呼び出します。
• fetch_greeting はリソースを読み込み、テキストだけを抽出します。
• assess_symptoms は簡潔なプロンプトを作成し、chat ツールを呼び出して、AI の応答を返します。

---

ステップ 2：ベーステンプレート（base.html）

すべてのページの壁紙とレイアウトを扱う単一のファイルです。

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <title>{% block title %}{% endblock %}</title>
  <link href="https://cdn.jsdelivr.net/npm/tailwindcss/dist/tailwind.min.css" rel="stylesheet">
  <style>
    body {
      background-image: url('{{ url_for("static", filename="assets/watsonx-wallpaper.jpg") }}');
      background-size: cover;
      background-position: center;
      background-repeat: no-repeat;
    }
  </style>
</head>
<body class="min-h-screen flex items-center justify-center p-4">
  {% block content %}{% endblock %}
</body>
</html>

説明：

• 壁紙を全画面の背景として読み込みます。
• Tailwind のユーティリティクラスを使ってコンテンツを中央に配置します。

---

ステップ 3：ホームページ（home.html）

{% extends "base.html" %}
{% block title %}Medical Chatbot – Welcome{% endblock %}
{% block content %}
  <div class="bg-white bg-opacity-80 p-8 rounded shadow-md w-full max-w-md">
    <h1 class="text-2xl mb-4 font-bold">Welcome to Your Medical Assistant</h1>
    <form method="post">
      <label class="block mb-2">What’s your name?</label>
      <input name="name" required class="w-full p-2 border rounded mb-4" placeholder="e.g. Angela"/>
      <button type="submit" class="w-full bg-blue-500 text-white py-2 rounded">Continue</button>
    </form>
  </div>
{% endblock %}

説明：

• base.html を継承して背景を引き継ぎます。
• name を収集するシンプルなフォームです。

---

ステップ 4：症状ページ（symptoms.html）

{% extends "base.html" %}
{% block title %}Medical Chatbot – Symptoms{% endblock %}
{% block content %}
  <div class="bg-white bg-opacity-80 p-8 rounded shadow-md w-full max-w-md">
    <h2 class="text-xl mb-4">{{ greeting }}</h2>
    <form method="post">
      <label class="block mb-2">Please describe your symptoms:</label>
      <textarea name="symptoms" rows="4" required class="w-full p-2 border rounded mb-4"
                placeholder="Headache, fever, coughing..."></textarea>
      <button type="submit" class="w-full bg-green-500 text-white py-2 rounded">Get Assessment</button>
    </form>
  </div>
{% endblock %}

説明：

• MCP から取得したあいさつを表示します。
• symptoms テキストを収集します。

---

ステップ 5：診断ページ（diagnosis.html）

{% extends "base.html" %}
{% block title %}Medical Chatbot – Assessment{% endblock %}
{% block content %}
  <div class="bg-white bg-opacity-80 p-8 rounded shadow-md w-full max-w-lg whitespace-pre-wrap">
    <h2 class="text-xl mb-4 font-bold">Assessment & Recommendations</h2>
    {{ diagnosis }}
    <div class="mt-6">
      <a 
        href="{{ url_for('home') }}" 
        class="w-full block bg-blue-500 text-white py-2 rounded text-center"
      >
        Start Over
      </a>
    </div>
  </div>
{% endblock %}

説明：

• 改行を保持したまま（whitespace-pre-wrap）、AI のアドバイスを描画します。

プロジェクト全体を作成したら、以下を実行します。

pyhon server.py 

別のターミナルで

python chatbot.py

すると、次のようになります。

次に、次の URL にアクセスします。 http://127.0.0.1:5000/

1. ホームページ

まず、患者の名前を入力します。

2. 症状ページ：

例として、次の症状を AI に入力できます。

Fever, Cough, Sore Throat, Muscle or Body Aches, Headache, Fatigue.

3. 診断ページ ここで AI は MCP サーバーを使用して WatsonX.ai に接続し、結果を取得します。

トラブルシューティングとベストプラクティス

• 「.env が読み込まれない」： os.getenv の前に load_dotenv() が呼び出されていることを確認してください。
• 「接続が拒否される」： クライアントより先にサーバー（mcp dev または python server.py）を実行してください。
• レイテンシーの急増： ストリーミングエンドポイントまたは、より小さなモデルの使用を検討してください。
• シークレットの管理： 本番環境では、.env に平文で保存する代わりに、安全なボルト（AWS Secrets Manager、IBM Key Protect）を使用してください。
• ログレベル： 開発中は DEBUG に切り替えてください（logging.basicConfig(level=logging.DEBUG)）。

---

次のステップと拡張

1. 会話状態： 最近の対話をリソースまたは永続化レイヤーに保存します。
2. プロンプトとテンプレート： @mcp.prompt() メソッドを追加して、あいさつのフローや FAQ を標準化します。
3. 追加のツール： ナレッジベース検索、感情分析、外部 API を統合します。
4. デプロイ： Docker でコンテナ化し、Kubernetes またはサーバーレスプラットフォームにデプロイします。
5. 認証： TLS またはトークンベースの認証を使用して、MCP の通信を保護します。

結論

おめでとうございます！ これで、MCP 経由で公開される、機能が充実した本番運用を意識した Watsonx.ai チャットボットサーバーが完成しました。このパターンでは、責務を明確に分離できます。

• MCP サーバー： ツールをホストし、ドキュメント化します
• Watsonx.ai： 高度な LLM 推論を処理します
• クライアント： 標準的な MCP コンシューマー（CLI、Web UI、デスクトップアプリ）

コーディングをお楽しみください！