PubMed MCP Server

Instructions

Implementation Reference

• src/tool_handler.py:728-809 (handler)

  Main execution logic for advanced_search tool: parses arguments, constructs complex PubMed query with field-specific terms and operators, applies filters (publication types, languages, species), performs search via pubmed_client, formats and returns results using _format_article_summary.

  async def _handle_advanced_search(self, arguments: Dict[str, Any]) -> MCPResponse:
      """Handle advanced search with multiple criteria."""
      try:
          search_terms = arguments.get("search_terms", [])
          if not search_terms:
              return MCPResponse(
                  content=[{"type": "text", "text": "Search terms are required"}], is_error=True
              )
  
          max_results = arguments.get("max_results", 50)
          filters = arguments.get("filters", {})
  
          # Build complex query
          query_parts = []
          for i, term_info in enumerate(search_terms):
              term = term_info.get("term", "")
              field = term_info.get("field", "all")
              operator = term_info.get("operator", "AND") if i > 0 else ""
  
              if field == "title":
                  field_query = f'"{term}"[Title]'
              elif field == "abstract":
                  field_query = f'"{term}"[Abstract]'
              elif field == "author":
                  field_query = f'"{term}"[Author]'
              elif field == "journal":
                  field_query = f'"{term}"[Journal]'
              elif field == "mesh":
                  field_query = f'"{term}"[MeSH Terms]'
              else:
                  field_query = term
  
              if i > 0:
                  query_parts.append(f" {operator} ")
              query_parts.append(f"({field_query})")
  
          query = "".join(query_parts)
  
          # Apply filters
          filter_parts = []
          if filters.get("publication_types"):
              type_queries = [f'"{pt}"[Publication Type]' for pt in filters["publication_types"]]
              filter_parts.append(f"({' OR '.join(type_queries)})")
  
          if filters.get("languages"):
              lang_queries = [f'"{lang}"[Language]' for lang in filters["languages"]]
              filter_parts.append(f"({' OR '.join(lang_queries)})")
  
          if filters.get("species"):
              if "humans" in [s.lower() for s in filters["species"]]:
                  filter_parts.append("humans[MeSH Terms]")
  
          if filter_parts:
              query += " AND " + " AND ".join(filter_parts)
  
          # Perform search
          search_result = await self.pubmed_client.search_articles(
              query=query, max_results=max_results, cache=self.cache
          )
  
          content = []
          content.append(
              {
                  "type": "text",
                  "text": f"**Advanced Search Results**\n\n"
                  f"Query: {query}\n"
                  f"Total Results: {search_result.total_results:,}\n"
                  f"Returned: {search_result.returned_results}\n",
              }
          )
  
          for i, article_data in enumerate(search_result.articles, 1):
              article_text = self._format_article_summary(article_data, i)
              content.append({"type": "text", "text": article_text})
  
          return MCPResponse(content=content)
  
      except Exception as e:
          logger.error(f"Error in advanced_search: {e}")
          return MCPResponse(
              content=[{"type": "text", "text": f"Error: {str(e)}"}], is_error=True
          )

• src/tools.py:319-356 (schema)

  Input schema defining parameters for advanced_search: array of search_terms (with term, field, operator), optional filters object, and max_results.

  {
      "name": "advanced_search",
      "description": ("Perform complex PubMed searches with multiple criteria"),
      "inputSchema": {
          "type": "object",
          "properties": {
              "search_terms": {
                  "type": "array",
                  "items": {
                      "type": "object",
                      "properties": {
                          "term": {"type": "string"},
                          "field": {
                              "type": "string",
                              "enum": ["title", "abstract", "author", "journal", "mesh", "all"],
                          },
                          "operator": {"type": "string", "enum": ["AND", "OR", "NOT"]},
                      },
                      "required": ["term", "field"],
                  },
                  "description": ("Complex search criteria with fields and operators"),
              },
              "filters": {
                  "type": "object",
                  "properties": {
                      "publication_types": {"type": "array", "items": {"type": "string"}},
                      "species": {"type": "array", "items": {"type": "string"}},
                      "languages": {"type": "array", "items": {"type": "string"}},
                      "age_groups": {"type": "array", "items": {"type": "string"}},
                      "sex": {"type": "string", "enum": ["male", "female", "both"]},
                  },
                  "description": "Additional filters",
              },
              "max_results": {"type": "integer", "minimum": 1, "maximum": 200, "default": 50},
          },
          "required": ["search_terms"],
      },
  },

• src/tool_handler.py:65-78 (registration)

  Maps tool name 'advanced_search' to its handler method _handle_advanced_search in the tool_handler's routing dictionary.

  handler_map = {
      "search_pubmed": self._handle_search_pubmed,
      "get_article_details": self._handle_get_article_details,
      "search_by_author": self._handle_search_by_author,
      "find_related_articles": self._handle_find_related_articles,
      "export_citations": self._handle_export_citations,
      "search_mesh_terms": self._handle_search_mesh_terms,
      "search_by_journal": self._handle_search_by_journal,
      "get_trending_topics": self._handle_get_trending_topics,
      "analyze_research_trends": self._handle_analyze_research_trends,
      "compare_articles": self._handle_compare_articles,
      "get_journal_metrics": self._handle_get_journal_metrics,
      "advanced_search": self._handle_advanced_search,
  }