使用说明

用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）推理，而模型上下文协议（MCP）标准化了应用程序如何向LLM客户端公开工具、资源和提示。通过将这两者结合起来，您可以得到：

模块化： 将聊天机器人逻辑与客户端实现解耦。
可重复使用性： 任何兼容MCP的客户端都可以调用相同的“聊天”端点。
快速迭代： 内置开发检查器，可实时重新加载。

无论您是在构建内部帮助台机器人程序还是公共聊天机器人程序API，此模式都可以轻松扩展和调整。

先决条件

在开始之前，请确保您已经：

IBM Cloud Watsonx.ai凭据： API密钥、服务URL和项目ID
Python 3.9+ （我们建议使用3.11+来改进性能和打字）
点（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聊天机器人服务器实现

环境设置

创建并激活虚拟环境

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

固定并安装依赖项

对于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

保护您的凭据
- 创建一个名为的文件 .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__/)to .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()

就是这样！您现在拥有：

A工作 chat 正确解析的工具 raw_response 来自Watsonx.ai。
A. 资源注入个性化问候。
可选提示医学症状评估模板。

运行和测试服务器

首先，我们加载环境

source .venv/bin/activate

开发模式与 `mcp dev`

最快的迭代方式是：

mcp dev server.py

实时重新加载 关于代码更改
检查器UI 在 http://localhost:6274/
交互式测试 chat 工具

如何与服务器交互

您可以进入 http://localhost:6274/

当您在浏览器中打开检查器时，您会注意到几个旨在促进服务器测试的关键部分。

转到MCP检查器界面的顶部，显示：

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

由于在我们的服务器中，.py是一个使用普通基于pip的虚拟环境的独立脚本，我们需要在MCP检查器中更正配置以：

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

然后单击Connect，我们的服务器将使用Python解释器正确启动。

然后转到工具，然后列出工具，在您可以键入的查询中单击聊天tna dhten

What is watsonx in IBM?

通过MCP检查员提示

在检查器的侧栏中，展开提示 → 列表提示.

你应该看看 assess_symptoms 上市的。

在提示窗格中，您将找到一个准备接受参数的表单或JSON编辑器。

供应 code 参数。例如：

persistent dull ache, stiffness, and general back pain

点击 获取提示.

您将收到：

{
  "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调用任何注册的资源：

打开“资源交互”窗格在MCP检查器侧栏中，单击资源→ 资源模板。单击“列出模板”，然后选择“绿化”
输入资源URI 在输入字段名称中，键入：约翰
调用资源单击“读取资源”。检查器将把该URI发送到您的@mcp.resource（“greeting://{name}”）处理程序。
查看响应您应该看到：

{
  "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客户端示例

将此另存为 client.py 旁边 server.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医疗聊天机器人

现在让我们使用以下方法创建一个基于网络的医疗聊天机器人：

MCP（模块化聊天协议） 构建我们的提示和工具调用
Watsonx.ai 用于LLM推理
烧瓶用于前端UI

到最后，你会有一个Flask应用程序：

问候按名称列出用户
收集症状
产生通过MCP进行诊断提示
呼叫 Watsonx.ai寻求医疗建议
显示风格化网页的评价

项目结构

├─ 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/ 目录。

第一步：写作 `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的回复。

第二步：基础模板(`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实用程序类来集中内容。

第三步：主页(`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/

首页

首先，您可以键入患者的姓名

症状页面：

您可以向AI提供以下症状作为示例

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

诊断页面在这里，AI将使用服务器MCP与WatsonX.AI连接，我们得到了结果。

故障排除和最佳实践

“.env未加载”： 确认 load_dotenv() 被调用之前 os.getenv.
“连接被拒绝”： 确保您正在运行服务器(mcp dev 或 python server.py)在客户之前。
延迟峰值： 考虑使用流端点或较小的模型。
秘密管理： 对于生产，请使用安全的保管库（AWS Secrets Manager、IBM Key Protect），而不是普通的 .env.
日志记录级别： 切换到 DEBUG 在开发过程中(logging.basicConfig(level=logging.DEBUG)).

后续步骤和扩展

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

结论

祝贺现在，您已经通过MCP公开了一个功能齐全、具有生产思维的Watsonx.ai聊天机器人服务器。此模式为您提供了清晰的关注点分离：

MCP服务器： 托管和记录您的工具
Watsonx.ai： 处理强大的LLM推理
客户： 任何标准MCP消费者（CLI、web UI、桌面应用程序）

编码愉快！

使用说明

用Python中的MCP构建Watsonx.ai聊天机器人服务器

引言

先决条件

项目结构

环境设置

编写聊天机器人服务器(server.py)

导入和配置

凭证验证和客户端初始化

定义MCP“聊天”工具

提示和资源

入口点和日志记录

运行和测试服务器

开发模式与 mcp dev

如何与服务器交互

通过MCP检查员提示

访问资源

直接执行

Python客户端示例

使用Flask和MCP构建Watsonx.ai医疗聊天机器人

项目结构

第一步：写作 chatbot.py

第二步：基础模板(base.html)

第三步：主页(home.html)

步骤4：症状页面(symptoms.html)

步骤5：诊断页面(diagnosis.html)

故障排除和最佳实践

后续步骤和扩展

结论

相关 MCP

Blender 3D建模MCP服务

iOS模拟器交互服务器

网页抓取与搜索MCP服务器

B站视频搜索服务

Minecraft MCP服务器

SSH远程命令执行MCP服务器

编写聊天机器人服务器(`server.py`)

开发模式与 `mcp dev`

第一步：写作 `chatbot.py`

第二步：基础模板(`base.html`)

第三步：主页(`home.html`)

步骤4：症状页面(`symptoms.html`)

步骤5：诊断页面(`diagnosis.html`)