mcp-scholar

Instructions

Implementation Reference

• src/mcp_scholar/server.py:321-370 (handler)

  MCP tool handler for the 'paper_references' tool. It calls the helper function get_paper_references and formats the response.
  @mcp.tool() async def paper_references( ctx: Context, paper_id: str, count: int = 5, sort_by: str = "relevance" ) -> Dict[str, Any]: """ 获取引用指定论文的文献列表 Args: paper_id: 论文ID count: 返回结果数量，默认为5 sort_by: 排序方式，可选值: - "relevance": 按相关性排序（默认） - "citations": 按引用量排序 - "date": 按发表日期排序（新到旧） - "title": 按标题字母顺序排序 Returns: Dict: 引用论文列表 """ try: # 移除进度显示 logger.info(f"正在获取论文ID为 {paper_id} 的引用...") references = await get_paper_references(paper_id, count, sort_by=sort_by) refs = [] for ref in references: refs.append( { "title": ref["title"], "authors": ref["authors"], "abstract": ref["abstract"], "citations": ref["citations"], "year": ref.get("year", "Unknown"), "paper_id": ref.get("paper_id", None), "url": ref.get("url", ""), # 添加URL "doi_url": ref.get("doi_url", ""), # 添加DOI URL } ) return { "status": "success", "references": refs, "sort_by": sort_by, } except Exception as e: error_msg = f"获取论文引用失败: {str(e)}" # 移除错误通知 logger.error(error_msg, exc_info=True) return {"status": "error", "message": "论文引用服务暂时不可用", "error": str(e)}
• src/mcp_scholar/scholar.py:338-458 (helper)

  Helper function implementing the core logic to fetch citing papers from OpenAlex API using the paper_id, handling different ID formats, sorting, and extracting relevant fields including abstract conversion from inverted index.
  async def get_paper_references( paper_id: str, count: int = 5, sort_by: str = "relevance" ) -> List[Dict[str, Any]]: """ 通过OpenAlex API获取引用指定论文的文献 Args: paper_id: 论文ID，可以是OpenAlex ID、DOI或ArXiv ID count: 返回结果数量 sort_by: 排序方式，可选值: - "relevance": 按相关性排序（默认） - "citations": 按引用量排序 - "date": 按发表日期排序（新到旧） - "title": 按标题字母顺序排序 Returns: List[Dict]: 引用论文信息列表 """ results = [] try: # 设置电子邮件参数（礼貌请求） email_param = f"&mailto={EMAIL}" if EMAIL else "" # 确定API ID openalex_id = paper_id # 如果是DOI或arXiv ID，需要先获取OpenAlex ID if paper_id.startswith("10.") or paper_id.lower().startswith("arxiv:"): id_type = "doi" if paper_id.startswith("10.") else "arxiv" id_value = ( paper_id if paper_id.startswith("10.") else paper_id.replace("arxiv:", "") ) # 查询以获取OpenAlex ID async with httpx.AsyncClient(timeout=10.0) as client: id_response = await client.get( f"{OPENALEX_API}/works/{id_type}:{id_value}?mailto={EMAIL if EMAIL else ''}" ) if id_response.status_code == 200: id_data = id_response.json() openalex_id = id_data.get("id", "").replace( "https://openalex.org/", "" ) else: print(f"获取OpenAlex ID错误: {id_response.status_code}") return [] # 去掉可能的前缀 if openalex_id.startswith("W"): openalex_id = openalex_id[1:] # 构建引用查询 citations_url = f"{OPENALEX_API}/works?filter=cites:W{openalex_id}{email_param}&per_page={count}" # 添加排序方式 if sort_by == "citations": citations_url += "&sort=cited_by_count:desc" elif sort_by == "date": citations_url += "&sort=publication_date:desc" elif sort_by == "title": citations_url += "&sort=title:asc" # 相关性(relevance)是默认排序，不需要额外参数 async with httpx.AsyncClient(timeout=15.0) as client: response = await client.get(citations_url) if response.status_code == 200: data = response.json() citation_papers = data.get("results", []) for paper_data in citation_papers: # 提取论文信息 paper = { "title": paper_data.get("title", "未知标题"), "abstract": "", # 默认为空，稍后处理 "citations": paper_data.get("cited_by_count", 0), "year": paper_data.get("publication_year", "未知年份"), "venue": "", # 需要从期刊/会议信息中提取 "paper_id": paper_data.get("id", "").replace( "https://openalex.org/", "" ), "url": paper_data.get("id", ""), } # 处理摘要（OpenAlex 摘要是倒排索引格式） if paper_data.get("abstract_inverted_index"): paper["abstract"] = convert_inverted_index_to_text( paper_data.get("abstract_inverted_index", {}) ) # 处理作者信息 authors = paper_data.get("authorships", []) author_names = [] for author in authors: if author.get("author", {}).get("display_name"): author_names.append(author["author"]["display_name"]) paper["authors"] = ", ".join(author_names) # 处理期刊/会议信息 if paper_data.get("host_venue", {}).get("display_name"): paper["venue"] = paper_data["host_venue"]["display_name"] # 处理DOI信息 if paper_data.get("doi"): paper["doi"] = paper_data["doi"] paper["doi_url"] = f"https://doi.org/{paper['doi']}" results.append(paper) return results else: print(f"获取论文引用错误: {response.status_code} - {response.text}") return [] except Exception as e: print(f"获取论文引用时出错: {str(e)}") return []