Watsonx MCP Server

用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聊天机器人服务器实现

环境设置

1. 创建并激活虚拟环境

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

1. 固定并安装依赖项

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

1. 保护您的凭据

- 创建一个名为的文件 .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()

就是这样！您现在拥有：

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

运行和测试服务器

首先，我们加载环境

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": "\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检查器侧栏中，单击资源→ 资源模板。单击“列出模板”，然后选择“绿化”

1. 输入资源URI

在输入字段名称中，键入： 约翰

1. 调用资源

单击“读取资源”。 检查器将把该URI发送到您的@mcp.resource（“greeting://{name}”）处理程序。

1. 查看响应

您应该看到：

{
  "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应用程序：

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

______________________________________________________________________

第一步：写作 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(" **说明：**
>
> - 我们打电话 `stdio_client` 和 `ClientSession` 在导入时，避免在每次HTTP请求时重新连接。
> - `fetch_greeting` 读取资源并仅提取文本。
> - `assess_symptoms` 构建一个干净的提示，调用 `chat` 工具，并返回AI的回复。

______________________________________________________________________

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

一个文件来处理所有页面的壁纸和布局。

{% block title %}{% endblock %}

body { background-image: url('{{ url_for("static", filename="assets/watsonx-wallpaper.jpg") }}'); background-size: cover; background-position: center; background-repeat: no-repeat; }

{% block content %}{% endblock %}

> **说明：**
>
> - 我们将壁纸加载为全屏背景。
> - 使用Tailwind实用程序类来集中内容。

______________________________________________________________________

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

{% extends "base.html" %} {% block title %}Medical Chatbot – Welcome{% endblock %} {% block content %}

Welcome to Your Medical Assistant

What’s your name?

Continue

{% endblock %}

> **说明：**
>
> - 扩展 `base.html` 继承背景。
> - 收集简单表格 `name`.

______________________________________________________________________

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

{% extends "base.html" %} {% block title %}Medical Chatbot – Symptoms{% endblock %} {% block content %}

{{ greeting }}

Please describe your symptoms:

Get Assessment

{% endblock %}

> **说明：**
>
> - 显示从MCP获取的问候语。
> - 收集 `symptoms` 文本。

______________________________________________________________________

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

{% extends "base.html" %} {% block title %}Medical Chatbot – Assessment{% endblock %} {% block content %}

Assessment & Recommendations

{{ diagnosis }}

Start Over

{% endblock %}

> **说明：**
>
> - 使用保留的换行符呈现AI的建议(`whitespace-pre-wrap`).

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

pyhon server.py

在另一个终端

python chatbot.py

你得到了
![](assets/2025-04-19-13-53-03.png)

然后您可以简单地进入
[http://127.0.0.1:5000/](http://127.0.0.1:5000/)

1. 首页

首先，您可以键入患者的姓名
![](assets/2025-04-19-13-35-13.png)

2. 症状页面：

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

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

![](assets/2025-04-19-13-36-05.png)

3. 诊断页面
   在这里，AI将使用服务器MCP与WatsonX.AI连接，我们得到了结果。
   ![](assets/2025-04-19-13-46-10.png)

## 故障排除和最佳实践

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

______________________________________________________________________

## 后续步骤和扩展

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

## 结论

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

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

编码愉快！