#!/usr/bin/env python3
"""
CNKI Search MCP Server
提供中文核心期刊论文检索能力的 MCP 服务器
使用免费的学术资源API
"""
import asyncio
import logging
from typing import Any
from mcp.server import Server
from mcp.types import Tool, TextContent
import requests
from urllib.parse import quote
import json
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 初始化 MCP 服务器
app = Server("cnki-search")
async def search_cnki_papers(arguments: dict) -> list[TextContent]:
"""搜索中文核心期刊论文"""
query = arguments.get("query", "")
max_results = arguments.get("max_results", 10)
year_start = arguments.get("year_start", "")
year_end = arguments.get("year_end", "")
if not query:
return [TextContent(type="text", text="Error: Query parameter is required")]
try:
# 使用百度学术API（免费）
url = "https://xueshu.baidu.com/s"
params = {
"wd": query,
"pn": 0,
"tn": "SE_baiduxueshu_c1gjeupa",
"ie": "utf-8",
"sc_hit": 1
}
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
# 由于百度学术需要解析HTML，我们使用备用方案：
# 使用开放的学术搜索API或构建简单的搜索
output = f"正在搜索中文核心期刊论文: '{query}'\n\n"
output += "注意：由于CNKI等数据库需要授权访问，本服务使用开放学术资源。\n"
output += "建议使用以下免费资源进行检索：\n\n"
output += "1. 百度学术: https://xueshu.baidu.com/\n"
output += "2. 谷歌学术: https://scholar.google.com/\n"
output += "3. 中国科技论文在线: http://www.paper.edu.cn/\n"
output += "4. 国家哲学社会科学文献中心: http://www.ncpssd.org/\n\n"
# 构建搜索链接
baidu_link = f"https://xueshu.baidu.com/s?wd={quote(query)}"
output += f"百度学术搜索链接: {baidu_link}\n"
return [TextContent(type="text", text=output)]
except Exception as e:
logger.error(f"Error searching papers: {e}")
return [TextContent(type="text", text=f"Error: {str(e)}")]
async def search_ncpssd_papers(arguments: dict) -> list[TextContent]:
"""搜索国家哲学社会科学文献中心论文（免费开放）"""
query = arguments.get("query", "")
max_results = arguments.get("max_results", 10)
if not query:
return [TextContent(type="text", text="Error: Query parameter is required")]
try:
output = f"搜索国家哲学社会科学文献中心: '{query}'\n\n"
output += "国家哲学社会科学文献中心提供免费的学术资源访问。\n\n"
# 构建搜索链接
search_link = f"http://www.ncpssd.org/literature/search.aspx?sw={quote(query)}"
output += f"搜索链接: {search_link}\n\n"
output += "该平台特点：\n"
output += "- 完全免费，无需注册\n"
output += "- 收录中文社科类核心期刊\n"
output += "- 提供全文下载\n"
output += "- 覆盖哲学、经济学、法学、社会学等领域\n"
return [TextContent(type="text", text=output)]
except Exception as e:
logger.error(f"Error searching NCPSSD: {e}")
return [TextContent(type="text", text=f"Error: {str(e)}")]
async def search_paper_edu_cn(arguments: dict) -> list[TextContent]:
"""搜索中国科技论文在线（免费开放）"""
query = arguments.get("query", "")
max_results = arguments.get("max_results", 10)
if not query:
return [TextContent(type="text", text="Error: Query parameter is required")]
try:
output = f"搜索中国科技论文在线: '{query}'\n\n"
output += "中国科技论文在线是教育部主办的免费学术平台。\n\n"
# 构建搜索链接
search_link = f"http://www.paper.edu.cn/search?q={quote(query)}"
output += f"搜索链接: {search_link}\n\n"
output += "该平台特点：\n"
output += "- 完全免费，开放获取\n"
output += "- 收录理工科类学术论文\n"
output += "- 提供首发论文和预印本\n"
output += "- 覆盖自然科学、工程技术等领域\n"
return [TextContent(type="text", text=output)]
except Exception as e:
logger.error(f"Error searching Paper.edu.cn: {e}")
return [TextContent(type="text", text=f"Error: {str(e)}")]
async def get_core_journal_info(arguments: dict) -> list[TextContent]:
"""获取中文核心期刊信息"""
journal_name = arguments.get("journal_name", "")
if not journal_name:
return [TextContent(type="text", text="Error: journal_name parameter is required")]
try:
output = f"查询核心期刊信息: '{journal_name}'\n\n"
output += "中文核心期刊分类：\n\n"
output += "1. 北大核心（中文核心期刊要目总览）\n"
output += " - 由北京大学图书馆编制\n"
output += " - 每4年更新一次\n"
output += " - 查询网站: http://hxqk.lib.pku.edu.cn/\n\n"
output += "2. CSSCI（中文社会科学引文索引）\n"
output += " - 南京大学主办\n"
output += " - 社科类权威索引\n"
output += " - 查询网站: http://cssci.nju.edu.cn/\n\n"
output += "3. 中国科技核心期刊\n"
output += " - 中国科技信息研究所发布\n"
output += " - 理工科类权威期刊\n\n"
return [TextContent(type="text", text=output)]
except Exception as e:
logger.error(f"Error getting journal info: {e}")
return [TextContent(type="text", text=f"Error: {str(e)}")]
@app.list_tools()
async def list_tools() -> list[Tool]:
"""列出可用的中文核心期刊检索工具"""
return [
Tool(
name="search_cnki_papers",
description="搜索中文核心期刊论文（提供免费学术资源链接）",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词（例如：'机器学习'、'量子计算'）"
},
"max_results": {
"type": "number",
"description": "最大返回结果数（默认：10）",
"default": 10
},
"year_start": {
"type": "string",
"description": "起始年份（可选）"
},
"year_end": {
"type": "string",
"description": "结束年份（可选）"
}
},
"required": ["query"]
}
),
Tool(
name="search_ncpssd_papers",
description="搜索国家哲学社会科学文献中心论文（免费开放，社科类）",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
},
"max_results": {
"type": "number",
"description": "最大返回结果数（默认：10）",
"default": 10
}
},
"required": ["query"]
}
),
Tool(
name="search_paper_edu_cn",
description="搜索中国科技论文在线（免费开放，理工科类）",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
},
"max_results": {
"type": "number",
"description": "最大返回结果数（默认：10）",
"default": 10
}
},
"required": ["query"]
}
),
Tool(
name="get_core_journal_info",
description="获取中文核心期刊分类信息",
inputSchema={
"type": "object",
"properties": {
"journal_name": {
"type": "string",
"description": "期刊名称（可选）"
}
},
"required": ["journal_name"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
"""处理工具调用"""
try:
if name == "search_cnki_papers":
return await search_cnki_papers(arguments)
elif name == "search_ncpssd_papers":
return await search_ncpssd_papers(arguments)
elif name == "search_paper_edu_cn":
return await search_paper_edu_cn(arguments)
elif name == "get_core_journal_info":
return await get_core_journal_info(arguments)
else:
raise ValueError(f"Unknown tool: {name}")
except Exception as e:
logger.error(f"Error calling tool {name}: {e}")
return [TextContent(type="text", text=f"Error: {str(e)}")]
async def main():
"""运行 MCP 服务器"""
from mcp.server.stdio import stdio_server
async with stdio_server() as (read_stream, write_stream):
await app.run(read_stream, write_stream, app.create_initialization_options())
if __name__ == "__main__":
asyncio.run(main())
