SuanDev/njupt-mcp

Fork 0

Files

MangoFanFanw 5ca759d280 init

2026-03-31 11:38:32 +08:00

5.4 KiB

Raw Permalink Blame History

开发指南

本文档介绍如何开发和扩展 NJUPT MCP Server。

架构概述

┌─────────────────┐
│   MCP Client    │  (Kimi Code CLI, Claude Desktop, etc.)
│  (Kimi, Claude) │
└────────┬────────┘
         │ MCP Protocol (stdio / SSE)
         ▼
┌─────────────────┐
│   njupt-mcp     │
│  (FastMCP)      │
└────────┬────────┘
         │
    ┌────┴────┐
    ▼         ▼
┌───────┐ ┌───────┐
│ Tools │ │Resources│
└───┬───┘ └───┬───┘
    │         │
    ▼         ▼
┌───────┐ ┌───────┐
│ NJUPT │ │ Static│
│ APIs  │ │ Data  │
└───────┘ └───────┘

添加新功能

1. 添加工具 (Tool)

工具用于执行操作、调用 API。在 src/njupt_mcp/server.py 中添加：

@mcp.tool()
async def my_tool(param1: str, param2: int = 10) -> str:
    """工具的描述，会显示给 LLM
    
    详细说明工具的用途、使用场景等。
    
    Args:
        param1: 参数1的描述
        param2: 参数2的描述，默认为 10
    
    Returns:
        返回值的描述，通常是 JSON 字符串
    
    Example:
        my_tool("test", 20) -> 返回结果
    """
    # 实现逻辑
    result = {"param1": param1, "param2": param2}
    return json.dumps(result, ensure_ascii=False)

最佳实践：

使用 async def 定义异步函数
提供详细的文档字符串（LLM 依赖此信息决定是否调用）
使用类型注解
返回 JSON 字符串便于 LLM 解析
处理异常情况，返回友好的错误信息

2. 添加资源 (Resource)

资源用于提供数据、文档等只读内容：

@mcp.resource("njupt://resource-name")
async def get_resource() -> str:
    """资源的描述"""
    return "资源内容"

资源 URI 格式：scheme://path，建议使用 njupt:// 作为 scheme。

3. 添加提示词模板 (Prompt)

提示词模板用于生成特定场景的提示词：

@mcp.prompt()
def my_prompt(context: str, question: str) -> str:
    """提示词模板的描述"""
    return f"""基于以下上下文：
{context}

请回答：{question}
"""

数据获取实现

目前示例代码使用硬编码数据，实际应用中需要对接 NJUPT 的数据源：

教务系统

import httpx

async def fetch_course_data(course_code: str) -> dict:
    """从教务系统获取课程数据"""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://jw.njupt.edu.cn/api/course",
            params={"code": course_code},
            cookies={"session": await get_session()}
        )
        response.raise_for_status()
        return response.json()

图书馆 OPAC

async def search_library(keyword: str) -> list[dict]:
    """搜索图书馆 OPAC 系统"""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "http://opac.njupt.edu.cn/opac/search",
            params={"q": keyword}
        )
        # 解析 HTML 或 JSON
        return parse_search_results(response.text)

测试

运行测试

# 运行所有测试
pytest

# 运行特定测试文件
pytest tests/test_helpers.py -v

# 运行并生成覆盖率报告
pytest --cov=njupt_mcp --cov-report=html

编写测试

# tests/test_new_feature.py
import pytest
from njupt_mcp.server import my_new_tool

@pytest.mark.asyncio
async def test_my_tool():
    result = await my_tool("test", 20)
    assert "test" in result
    assert "20" in result

调试技巧

使用 MCP Inspector

# 启动服务器
uv run njupt-mcp --transport streamable-http

# 启动 Inspector
npx -y @modelcontextprotocol/inspector

访问 http://localhost:5173 进行交互式调试。

启用调试日志

uv run njupt-mcp --debug

手动测试工具

# test_manual.py
import asyncio
from njupt_mcp.server import search_course

async def main():
    result = await search_course("数据结构")
    print(result)

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

部署

本地部署

使用 uv 运行：

uv run njupt-mcp

Docker 部署

FROM python:3.11-slim

WORKDIR /app
COPY . .

RUN pip install -e "."

EXPOSE 8000

CMD ["njupt-mcp", "--transport", "sse", "--host", "0.0.0.0"]

云服务部署

适合使用 SSE 或 streamable-http 模式，部署到：

性能优化

添加缓存

from functools import lru_cache
import asyncio
from cachetools import TTLCache

# 使用 TTLCache 缓存结果
course_cache = TTLCache(maxsize=100, ttl=3600)

@mcp.tool()
async def search_course(keyword: str) -> str:
    if keyword in course_cache:
        return course_cache[keyword]
    
    result = await fetch_course_from_api(keyword)
    course_cache[keyword] = result
    return result

并发请求

async def fetch_multiple_courses(course_codes: list[str]) -> list[dict]:
    async with httpx.AsyncClient() as client:
        tasks = [fetch_course(client, code) for code in course_codes]
        return await asyncio.gather(*tasks)

5.4 KiB Raw Permalink Blame History Unescape Escape