看护资源MCP服务器

一个基于FastAPI的MCP服务器,通过Server-Sent Events (SSE)提供来自AARP.org的看护资源。

  • by rmiranda-dev
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 rmiranda-dev/caregiving-mcp

使用说明

护理管理控制平台(MCP)服务器

一个基于FastAPI的MCP(模型上下文协议)服务器,通过服务器发送事件(SSE)从AARP.org提供照护资源。

特点/特性

  • **MCP-over-SSE(MCP通过SSE传输)**流媒体终端位于 /sse 用于实时工具执行
  • 两个MCP工具:
    • search_caregiving(query, limit=5)在AARP.org上搜索有关护理的文章
    • fetch_article(url)获取并提取文章内容(限于AARP(美国退休人员协会)护理领域)
  • 携带令牌认证可选的基于令牌的认证通过 MCP_TOKEN 环境变量
  • 健康检查简单 / 端点返回服务状态
  • 内容限制文章文本自动限制在约2万字符以内

快速入门

1. 创建虚拟环境

# Windows PowerShell
python -m venv venv
.\venv\Scripts\activate
# Linux/Mac
python3 -m venv venv
source venv/bin/activate

2. 安装依赖项

pip install -r requirements.txt

3. 启动服务器

基本(无需认证):

python main.py

带认证:

# Windows PowerShell
$env:MCP_TOKEN="your-secret-token"
python main.py
# Linux/Mac
export MCP_TOKEN="your-secret-token"
python main.py

自定义端口:

# Windows PowerShell
$env:PORT="3000"
python main.py
# Linux/Mac
export PORT=3000
python main.py

服务器将在 http://localhost:8000 (或您的自定义端口)

API终端点

健康检查

GET /

返回值:

{
  "status": "healthy",
  "service": "caregiving-mcp",
  "version": "1.0.0"
}

MCP-over-SSE 端点

POST /sse
Authorization: Bearer your-secret-token (if MCP_TOKEN is set)

这个端点通过服务器发送事件(Server-Sent Events)处理MCP协议消息。

MCP 工具

搜索护理服务

在AARP.org上搜索有关护理的文章。

参数:

  • query (字符串,必填):搜索查询
  • limit (整数,可选):最大结果数(默认:5)

返回值: 包含标题、网址和摘要的搜索结果JSON

获取文章

从AARP的护理文章中抓取并提取内容。

参数:

  • url (字符串,必填):文章URL(必须来自https://www.aarp.org/caregiving/)

返回: 文章内容以格式化文本形式呈现(字符数上限约为2万)

安全

  • URLs(统一资源定位符)用于 fetch_article 被限制在……之内 https://www.aarp.org/caregiving/ 域名(或领域、范围)
  • 可选的承载令牌认证 /sse 终端节点
  • 配置了30秒超时的HTTPX客户端,以防止请求挂起

依赖项

  • FastAPI用于构建API的现代Web框架
  • Uvicorn用于运行 FastAPI 的 ASGI 服务器
  • **httpx(注:httpx通常指的是一种用于发送HTTP请求的工具或库,但在此上下文中,由于没有具体语境,仅将其作为名词直译为“httpx”)**用于网页抓取的异步HTTP客户端
  • **选择性泻药(或译为:选性通便剂,但“selectolax”并非标准医学术语,此处为意译)**快速HTML解析器,用于内容提取
  • fastmcp支持SSE的轻量级MCP协议实现

发展

以开发模式运行并启用自动重载:

uvicorn main:app --reload --port 8000

MCP客户端使用示例

import httpx

async def test_mcp_tools():
    async with httpx.AsyncClient() as client:
        # Health check
        response = await client.get("http://localhost:8000/")
        print(response.json())
        
        # Connect to SSE endpoint
        headers = {"Authorization": "Bearer your-secret-token"}  # if auth enabled
        async with client.stream("POST", "http://localhost:8000/sse", headers=headers) as response:
            async for line in response.aiter_lines():
                print(line)

许可证

MIT 许可证 - 欢迎自由使用并根据需要进行修改。