init
This commit is contained in:
12
src/njupt_mcp/__init__.py
Normal file
12
src/njupt_mcp/__init__.py
Normal file
@@ -0,0 +1,12 @@
|
||||
"""NJUPT MCP Server - 南京邮电大学 MCP 服务器
|
||||
|
||||
为 LLM 提供获取南京邮电大学相关信息的能力。
|
||||
"""
|
||||
|
||||
__version__ = "0.1.0"
|
||||
__author__ = "Your Name"
|
||||
__description__ = "南京邮电大学 (NJUPT) MCP 服务器"
|
||||
|
||||
from .server import mcp
|
||||
|
||||
__all__ = ["mcp"]
|
||||
8
src/njupt_mcp/resources/__init__.py
Normal file
8
src/njupt_mcp/resources/__init__.py
Normal file
@@ -0,0 +1,8 @@
|
||||
"""Resources 模块
|
||||
|
||||
包含各种资源定义,用于提供静态或动态的数据资源。
|
||||
|
||||
资源与工具的区别:
|
||||
- 工具 (Tools): 执行操作、调用 API、产生副作用
|
||||
- 资源 (Resources): 提供数据、文档、上下文信息,通常是只读的
|
||||
"""
|
||||
376
src/njupt_mcp/server.py
Normal file
376
src/njupt_mcp/server.py
Normal file
@@ -0,0 +1,376 @@
|
||||
"""NJUPT MCP Server 主入口
|
||||
|
||||
基于 FastMCP 实现的南京邮电大学 MCP 服务器。
|
||||
支持 stdio 和 SSE 两种传输方式。
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import logging
|
||||
import sys
|
||||
from contextlib import asynccontextmanager
|
||||
from typing import AsyncIterator
|
||||
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
from mcp.server.fastmcp.server import Settings
|
||||
|
||||
# 配置日志
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||||
)
|
||||
logger = logging.getLogger("njupt-mcp")
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def app_lifespan(server: FastMCP) -> AsyncIterator[dict]:
|
||||
"""应用生命周期管理
|
||||
|
||||
处理服务器启动和关闭时的资源初始化和清理。
|
||||
"""
|
||||
logger.info("🚀 NJUPT MCP Server 启动中...")
|
||||
|
||||
# 初始化共享资源(如数据库连接、HTTP 客户端等)
|
||||
try:
|
||||
# 这里可以初始化一些全局资源
|
||||
lifespan_context = {
|
||||
"server_name": "njupt-mcp",
|
||||
"version": "0.1.0",
|
||||
}
|
||||
logger.info("✅ 服务器启动成功")
|
||||
yield lifespan_context
|
||||
finally:
|
||||
logger.info("🛑 NJUPT MCP Server 关闭中...")
|
||||
# 清理资源
|
||||
logger.info("✅ 资源清理完成")
|
||||
|
||||
|
||||
# 创建 MCP 服务器实例
|
||||
# lifespan: 应用生命周期管理
|
||||
mcp = FastMCP(
|
||||
"njupt-mcp",
|
||||
lifespan=app_lifespan,
|
||||
)
|
||||
|
||||
|
||||
# ==================== Tools ====================
|
||||
|
||||
@mcp.tool()
|
||||
async def search_course(keyword: str, limit: int = 10) -> str:
|
||||
"""搜索南京邮电大学课程信息
|
||||
|
||||
根据关键词搜索课程名称、课程代码、开课学院等信息。
|
||||
|
||||
Args:
|
||||
keyword: 搜索关键词(课程名称、课程代码等)
|
||||
limit: 返回结果数量限制,默认 10 条
|
||||
|
||||
Returns:
|
||||
课程信息列表的 JSON 字符串
|
||||
|
||||
Example:
|
||||
search_course("数据结构") -> 返回数据结构相关课程
|
||||
search_course("CS101", limit=5) -> 返回课程代码包含 CS101 的课程
|
||||
"""
|
||||
# TODO: 实现实际的课程搜索逻辑
|
||||
# 这里返回示例数据
|
||||
courses = [
|
||||
{
|
||||
"course_code": "CS2101",
|
||||
"name": f"{keyword}基础",
|
||||
"credit": 3.0,
|
||||
"department": "计算机学院",
|
||||
"description": f"这是一门关于{keyword}的基础课程",
|
||||
},
|
||||
{
|
||||
"course_code": "CS3102",
|
||||
"name": f"高级{keyword}",
|
||||
"credit": 4.0,
|
||||
"department": "计算机学院",
|
||||
"description": f"深入探讨{keyword}的高级主题",
|
||||
},
|
||||
]
|
||||
|
||||
import json
|
||||
return json.dumps(courses[:limit], ensure_ascii=False, indent=2)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def get_course_schedule(student_id: str, semester: str = "2024-2025-1") -> str:
|
||||
"""获取学生课表信息
|
||||
|
||||
根据学号和学期获取个人课表。
|
||||
|
||||
Args:
|
||||
student_id: 学生学号(如 B21010101)
|
||||
semester: 学期代码,格式为 "YYYY-YYYY-S"(如 2024-2025-1)
|
||||
|
||||
Returns:
|
||||
课表信息的 JSON 字符串
|
||||
|
||||
Example:
|
||||
get_course_schedule("B21010101", "2024-2025-1")
|
||||
"""
|
||||
# TODO: 实现实际的课表查询逻辑
|
||||
import json
|
||||
schedule = {
|
||||
"student_id": student_id,
|
||||
"semester": semester,
|
||||
"courses": [
|
||||
{
|
||||
"day": 1,
|
||||
"period": [1, 2],
|
||||
"name": "数据结构",
|
||||
"location": "教1-101",
|
||||
"teacher": "张三教授",
|
||||
},
|
||||
{
|
||||
"day": 3,
|
||||
"period": [3, 4],
|
||||
"name": "计算机网络",
|
||||
"location": "教2-205",
|
||||
"teacher": "李四副教授",
|
||||
},
|
||||
],
|
||||
}
|
||||
return json.dumps(schedule, ensure_ascii=False, indent=2)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def search_library_book(keyword: str, search_type: str = "title") -> str:
|
||||
"""搜索图书馆藏书
|
||||
|
||||
在南京邮电大学图书馆搜索图书。
|
||||
|
||||
Args:
|
||||
keyword: 搜索关键词
|
||||
search_type: 搜索类型,可选 "title"(书名), "author"(作者), "isbn"(ISBN)
|
||||
|
||||
Returns:
|
||||
图书信息列表的 JSON 字符串
|
||||
|
||||
Example:
|
||||
search_library_book("Python", "title")
|
||||
search_library_book("鲁迅", "author")
|
||||
"""
|
||||
# TODO: 实现实际的图书馆搜索逻辑
|
||||
import json
|
||||
books = [
|
||||
{
|
||||
"title": f"{keyword}编程实战",
|
||||
"author": "王某某",
|
||||
"publisher": "清华大学出版社",
|
||||
"isbn": "978-7-302-12345-6",
|
||||
"available": True,
|
||||
"location": "计算机科学阅览室",
|
||||
},
|
||||
{
|
||||
"title": f"{keyword}入门指南",
|
||||
"author": "李某某",
|
||||
"publisher": "人民邮电出版社",
|
||||
"isbn": "978-7-115-78901-2",
|
||||
"available": False,
|
||||
"location": "基础科学阅览室",
|
||||
},
|
||||
]
|
||||
return json.dumps(books, ensure_ascii=False, indent=2)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def get_campus_info(info_type: str = "overview") -> str:
|
||||
"""获取校园信息
|
||||
|
||||
获取南京邮电大学的基本信息、办事指南等。
|
||||
|
||||
Args:
|
||||
info_type: 信息类型,可选 "overview"(概况), "map"(地图),
|
||||
"contacts"(联系方式), "calendar"(校历)
|
||||
|
||||
Returns:
|
||||
校园信息的字符串
|
||||
|
||||
Example:
|
||||
get_campus_info("overview") -> 学校概况
|
||||
get_campus_info("contacts") -> 各部门联系方式
|
||||
"""
|
||||
info_data = {
|
||||
"overview": """南京邮电大学(Nanjing University of Posts and Telecommunications,
|
||||
简称 NJUPT)是国家“双一流”建设高校和江苏高水平大学高峰计划 A 类建设高校。
|
||||
学校坐落于历史文化名城南京,现有仙林、三牌楼、锁金村、江宁四个校区。""",
|
||||
"contacts": """教务处:025-85866250
|
||||
学生工作处:025-85866255
|
||||
图书馆:025-85866270
|
||||
信息化建设与管理办公室:025-85866280""",
|
||||
"map": "请访问 https://map.njupt.edu.cn 查看校园地图",
|
||||
"calendar": "2024-2025学年第一学期:2024年9月2日 - 2025年1月12日",
|
||||
}
|
||||
return info_data.get(info_type, info_data["overview"])
|
||||
|
||||
|
||||
# ==================== Resources ====================
|
||||
|
||||
@mcp.resource("njupt://announcements")
|
||||
async def get_announcements() -> str:
|
||||
"""获取南京邮电大学最新公告
|
||||
|
||||
Returns:
|
||||
最新公告列表
|
||||
"""
|
||||
# TODO: 实现实际的公告获取逻辑
|
||||
announcements = """📢 南京邮电大学最新公告:
|
||||
|
||||
1. 【教务处】关于2024-2025学年第一学期期末考试安排的通知
|
||||
2. 【图书馆】寒假期间图书馆开放时间安排
|
||||
3. 【学生处】关于评选2024年度优秀学生奖学金的通知
|
||||
4. 【信息化办】校园网络升级维护公告
|
||||
|
||||
更多公告请访问 https://www.njupt.edu.cn"""
|
||||
return announcements
|
||||
|
||||
|
||||
@mcp.resource("njupt://academic-calendar")
|
||||
async def get_academic_calendar() -> str:
|
||||
"""获取南京邮电大学校历
|
||||
|
||||
Returns:
|
||||
当前学年校历信息
|
||||
"""
|
||||
return """📅 2024-2025学年校历:
|
||||
|
||||
第一学期:
|
||||
- 开学:2024年9月2日
|
||||
- 教学周:第1-18周(9月2日 - 1月5日)
|
||||
- 考试周:第19-20周(1月6日 - 1月12日)
|
||||
- 寒假:2025年1月13日 - 2月16日
|
||||
|
||||
第二学期:
|
||||
- 开学:2025年2月17日
|
||||
- 教学周:第1-18周(2月17日 - 6月22日)
|
||||
- 考试周:第19-20周(6月23日 - 6月29日)
|
||||
- 暑假:2025年6月30日起"""
|
||||
|
||||
|
||||
@mcp.resource("njupt://departments")
|
||||
async def get_departments() -> str:
|
||||
"""获取南京邮电大学学院/部门列表
|
||||
|
||||
Returns:
|
||||
学院和部门列表
|
||||
"""
|
||||
return """🏫 南京邮电大学学院设置:
|
||||
|
||||
通信与信息工程学院
|
||||
电子与光学工程学院/柔性电子(未来技术)学院
|
||||
集成电路科学与工程学院(产教融合学院)
|
||||
计算机学院/软件学院/网络空间安全学院
|
||||
自动化学院/人工智能学院
|
||||
材料科学与工程学院
|
||||
化学与生命科学学院
|
||||
物联网学院
|
||||
理学院
|
||||
地理与生物信息学院
|
||||
现代邮政学院
|
||||
传媒与艺术学院
|
||||
管理学院
|
||||
经济学院
|
||||
马克思主义学院
|
||||
社会与人口学院/社会工作学院
|
||||
外国语学院
|
||||
教育科学与技术学院
|
||||
|
||||
📋 主要职能部门:
|
||||
教务处、学生工作处、研究生工作部、科学技术处、
|
||||
人事处、财务处、审计处、保卫处、后勤管理处等"""
|
||||
|
||||
|
||||
# ==================== Prompts ====================
|
||||
|
||||
@mcp.prompt()
|
||||
def academic_advisor_query(question: str, student_major: str = "") -> str:
|
||||
"""学业咨询助手 Prompt
|
||||
|
||||
生成针对学业问题的咨询提示词。
|
||||
|
||||
Args:
|
||||
question: 学生的问题
|
||||
student_major: 学生专业(可选)
|
||||
"""
|
||||
major_context = f"该学生专业为:{student_major}。" if student_major else ""
|
||||
|
||||
return f"""你是南京邮电大学的学业咨询助手,专门帮助学生解决学习和课程相关的问题。
|
||||
|
||||
{major_context}
|
||||
|
||||
学生问题:{question}
|
||||
|
||||
请以专业、友善的态度回答,并提供具体可行的建议。如果涉及课程选择、学分计算等问题,请给出详细的说明。"""
|
||||
|
||||
|
||||
@mcp.prompt()
|
||||
def campus_guide_query(location: str, query_type: str = "location") -> str:
|
||||
"""校园导航助手 Prompt
|
||||
|
||||
生成校园导航相关的提示词。
|
||||
|
||||
Args:
|
||||
location: 地点名称或描述
|
||||
query_type: 查询类型,可选 "location"(位置), "route"(路线), "facility"(设施)
|
||||
"""
|
||||
prompts = {
|
||||
"location": f"请介绍南京邮电大学'{location}'的具体位置和周边信息。",
|
||||
"route": f"请提供前往南京邮电大学'{location}'的详细路线指引。",
|
||||
"facility": f"请介绍南京邮电大学'{location}'的设施情况和开放时间。",
|
||||
}
|
||||
|
||||
base_prompt = prompts.get(query_type, prompts["location"])
|
||||
|
||||
return f"""你是南京邮电大学校园导航助手,熟悉校园的各个位置和设施。
|
||||
|
||||
{base_prompt}
|
||||
|
||||
请提供准确、详细的信息,必要时可以提及周边的其他相关地点。"""
|
||||
|
||||
|
||||
def main():
|
||||
"""命令行入口点"""
|
||||
parser = argparse.ArgumentParser(description="NJUPT MCP Server")
|
||||
parser.add_argument(
|
||||
"--transport",
|
||||
choices=["stdio", "sse", "streamable-http"],
|
||||
default="stdio",
|
||||
help="传输协议类型 (默认: stdio)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--host",
|
||||
default="127.0.0.1",
|
||||
help="SSE/HTTP 模式下的监听地址 (默认: 127.0.0.1)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--port",
|
||||
type=int,
|
||||
default=8000,
|
||||
help="SSE/HTTP 模式下的监听端口 (默认: 8000)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--debug",
|
||||
action="store_true",
|
||||
help="启用调试模式",
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.debug:
|
||||
logging.getLogger().setLevel(logging.DEBUG)
|
||||
logger.debug("调试模式已启用")
|
||||
|
||||
logger.info(f"启动 NJUPT MCP Server,传输方式: {args.transport}")
|
||||
|
||||
if args.transport == "stdio":
|
||||
mcp.run(transport="stdio")
|
||||
elif args.transport == "sse":
|
||||
mcp.run(transport="sse", host=args.host, port=args.port)
|
||||
elif args.transport == "streamable-http":
|
||||
mcp.run(transport="streamable-http", host=args.host, port=args.port)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
12
src/njupt_mcp/tools/__init__.py
Normal file
12
src/njupt_mcp/tools/__init__.py
Normal file
@@ -0,0 +1,12 @@
|
||||
"""Tools 模块
|
||||
|
||||
包含各种工具函数,用于实现与南京邮电大学相关的功能。
|
||||
|
||||
每个工具函数应该:
|
||||
1. 有清晰的文档字符串说明用途和参数
|
||||
2. 使用类型注解
|
||||
3. 返回结构化的数据(通常是 JSON 字符串)
|
||||
4. 处理异常情况并返回友好的错误信息
|
||||
"""
|
||||
|
||||
# 工具函数将在这里定义,也可以在单独的模块中定义后在此导入
|
||||
8
src/njupt_mcp/utils/__init__.py
Normal file
8
src/njupt_mcp/utils/__init__.py
Normal file
@@ -0,0 +1,8 @@
|
||||
"""Utils 模块
|
||||
|
||||
包含通用的工具函数和辅助类。
|
||||
"""
|
||||
|
||||
from .helpers import format_json, parse_semester
|
||||
|
||||
__all__ = ["format_json", "parse_semester"]
|
||||
79
src/njupt_mcp/utils/helpers.py
Normal file
79
src/njupt_mcp/utils/helpers.py
Normal file
@@ -0,0 +1,79 @@
|
||||
"""通用工具函数"""
|
||||
|
||||
import json
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
|
||||
def format_json(data: Any, ensure_ascii: bool = False, indent: int = 2) -> str:
|
||||
"""将数据格式化为 JSON 字符串
|
||||
|
||||
Args:
|
||||
data: 要格式化的数据
|
||||
ensure_ascii: 是否确保 ASCII 编码(False 支持中文)
|
||||
indent: 缩进空格数
|
||||
|
||||
Returns:
|
||||
格式化后的 JSON 字符串
|
||||
"""
|
||||
return json.dumps(data, ensure_ascii=ensure_ascii, indent=indent)
|
||||
|
||||
|
||||
def parse_semester(semester_str: str) -> tuple[str, str, int] | None:
|
||||
"""解析学期字符串
|
||||
|
||||
格式: "YYYY-YYYY-S"(如 "2024-2025-1")
|
||||
|
||||
Args:
|
||||
semester_str: 学期字符串
|
||||
|
||||
Returns:
|
||||
(起始年份, 结束年份, 学期编号) 或 None(解析失败)
|
||||
|
||||
Example:
|
||||
>>> parse_semester("2024-2025-1")
|
||||
("2024", "2025", 1)
|
||||
"""
|
||||
pattern = r"^(\d{4})-(\d{4})-(\d)$"
|
||||
match = re.match(pattern, semester_str)
|
||||
|
||||
if match:
|
||||
start_year, end_year, semester_num = match.groups()
|
||||
return start_year, end_year, int(semester_num)
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def validate_student_id(student_id: str) -> bool:
|
||||
"""验证学号格式
|
||||
|
||||
南邮学号格式:
|
||||
- 本科:B + 年级(2位) + 学院代码(2位) + 班级(2位) + 序号(2位)
|
||||
如 B21010101
|
||||
- 研究生:其他格式
|
||||
|
||||
Args:
|
||||
student_id: 学号字符串
|
||||
|
||||
Returns:
|
||||
是否有效
|
||||
"""
|
||||
# 本科学号正则
|
||||
undergraduate_pattern = r"^[BM]\d{8}$"
|
||||
return bool(re.match(undergraduate_pattern, student_id))
|
||||
|
||||
|
||||
def sanitize_input(text: str) -> str:
|
||||
"""清理用户输入,防止 XSS 等攻击
|
||||
|
||||
Args:
|
||||
text: 输入字符串
|
||||
|
||||
Returns:
|
||||
清理后的字符串
|
||||
"""
|
||||
# 移除 HTML 标签
|
||||
text = re.sub(r"<[^>]*>", "", text)
|
||||
# 移除特殊字符
|
||||
text = text.replace("\x00", "")
|
||||
return text.strip()
|
||||
Reference in New Issue
Block a user