Building a Watsonx.ai Chatbot Server with MCP in Python

使用 Python 和 MCP 构建 Watsonx.ai 聊天机器人服务

在这篇深入教程中，你将学习如何创建一个专业、可用于生产环境的聊天机器人服务，该服务由 IBM Watsonx.ai 提供支持，并通过 MCP Python SDK 暴露。完成后，你将拥有一个可复用的 MCP 服务，任何兼容 MCP 的客户端（例如 Claude Desktop、自定义 Python 客户端）都可以将其作为标准化的“聊天”工具调用。

我们将逐步介绍：

• 一步步设置环境
• 安装依赖项并安全管理凭据
• 编写清晰、文档完善的 Python 代码
• 将 Watsonx.ai 推理功能暴露为 MCP 工具
• 运行、测试和排查服务问题
• 扩展和加固服务的技巧

简介

IBM Watsonx.ai 通过 IBM Cloud 提供先进的大模型（LLM）推理功能，而 Model Context Protocol（MCP）则标准化了应用向大模型客户端暴露工具、资源和提示词的方式。将两者结合后，你将获得：

• 模块化： 将聊天机器人逻辑与客户端实现解耦。
• 可复用性： 任何兼容 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}"

---

提示词与资源

暴露资源
资源可让你将动态数据注入大模型的上下文。例如，我们在此暴露一条个性化问候语：

# 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. 一个可正常工作的 chat 工具，能够正确解析 Watsonx.ai 返回的 raw_response。
2. 一个用于注入个性化问候语的资源。
3. 一个用于评估医疗症状的可选提示词模板。

运行和测试服务

首先加载环境

source .venv/bin/activate

使用 mcp dev 进行开发

最快的迭代方式是：

mcp dev server.py

• 代码变更时实时重新加载
• http://localhost:6274/ 上的检查器 UI
• 交互式测试 chat 工具

如何与服务交互

你可以访问 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 使用提示词

在 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."
      }
    }
  ]
}

访问资源

在 Inspector 中启动并运行服务（使用上述设置）后，你可以通过 URI 调用任何已注册的资源：

1. 打开资源交互面板 在 MCP Inspector 侧边栏中，点击 Resources → Resource Templates。点击 List templates，然后选择 get_greeting。
2. 输入资源 URI 在输入字段中输入： John
3. 调用资源 点击 Read Resource。 Inspector 会将该 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?"
    }
  ]
}

这表明你的动态问候资源已正确配置，并能按需返回个性化内容。

直接执行

若要在不使用 Inspector 的情况下运行：

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）**来组织提示词和工具调用
• Watsonx.ai 进行大模型推理
• Flask 构建前端 UI

完成后，你将拥有一个 Flask 应用，它可以：

1. 按姓名问候用户
2. 收集用户的症状
3. 通过 MCP 生成诊断提示词
4. 调用 Watsonx.ai 获取医疗建议
5. 在经过样式设计的网页上显示评估结果

项目结构

├─ 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 客户端会话，并在多个请求间重复使用。

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)

说明：

• 我们在导入时调用 stdio_client 和 ClientSession，以避免每次 HTTP 请求都重新连接。
• 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 %}

说明：

• 呈现 AI 的建议，并保留换行符（whitespace-pre-wrap）。

创建完整项目后，只需运行

pyhon server.py 

然后在另一个终端中运行

python chatbot.py

你会看到

然后只需访问 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）。
• 延迟激增： 考虑使用流式端点或更小的模型。
• 密钥管理： 在生产环境中，使用安全的密钥库（AWS Secrets Manager、IBM Key Protect），而不是明文 .env。
• 日志级别： 开发期间将级别切换为 DEBUG（logging.basicConfig(level=logging.DEBUG)）。

---

后续步骤与扩展

1. 对话状态： 将最近的对话存储在资源或持久化层中。
2. 提示词与模板： 添加 @mcp.prompt() 方法，以规范问候流程或常见问题解答。
3. 其他工具： 集成知识库搜索、情感分析或外部 API。
4. 部署： 使用 Docker 容器化，并部署到 Kubernetes 或无服务器平台。
5. 身份验证： 使用 TLS 或基于令牌的身份验证来保护 MCP 传输。

结语

恭喜！ 现在，你已经拥有一个功能齐全、以生产环境为目标的 Watsonx.ai 聊天机器人服务，并通过 MCP 对外提供。此模式清晰地分离了各项职责：

• MCP 服务： 托管并记录你的工具
• Watsonx.ai： 处理强大的大模型推理
• 客户端： 任何标准的 MCP 使用方（命令行、Web UI、桌面应用）

祝编码愉快！