Microsoft MCP

Instructions

entity_types can include: 'message', 'event', 'drive', 'driveItem', 'list', 'listItem', 'site'
If not specified, searches across all available types.

Implementation Reference

• src/microsoft_mcp/tools.py:941-973 (handler)

  The main handler function for the 'unified_search' tool. It accepts a query, account_id, optional entity_types and limit, calls graph.search_query, categorizes results by entity type (message, event, driveItem, other), and returns a dictionary of lists grouped by type.

  @mcp.tool
  def unified_search(
      query: str,
      account_id: str,
      entity_types: list[str] | None = None,
      limit: int = 50,
  ) -> dict[str, list[dict[str, Any]]]:
      """Search across multiple Microsoft 365 resources using the modern search API
  
      entity_types can include: 'message', 'event', 'drive', 'driveItem', 'list', 'listItem', 'site'
      If not specified, searches across all available types.
      """
      if not entity_types:
          entity_types = ["message", "event", "driveItem"]
  
      results = {entity_type: [] for entity_type in entity_types}
  
      items = list(graph.search_query(query, entity_types, account_id, limit))
  
      for item in items:
          resource_type = item.get("@odata.type", "").split(".")[-1]
  
          if resource_type == "message":
              results.setdefault("message", []).append(item)
          elif resource_type == "event":
              results.setdefault("event", []).append(item)
          elif resource_type in ["driveItem", "file", "folder"]:
              results.setdefault("driveItem", []).append(item)
          else:
              results.setdefault("other", []).append(item)
  
      return {k: v for k, v in results.items() if v}

• src/microsoft_mcp/graph.py:277-331 (helper)

  Supporting utility function 'search_query' that implements the Microsoft Graph /search/query API call with pagination handling via 'from' parameter and yields search hit resources. Directly called by unified_search.

  def search_query(
      query: str,
      entity_types: list[str],
      account_id: str | None = None,
      limit: int = 50,
      fields: list[str] | None = None,
  ) -> Iterator[dict[str, Any]]:
      """Use the modern /search/query API endpoint"""
      payload = {
          "requests": [
              {
                  "entityTypes": entity_types,
                  "query": {"queryString": query},
                  "size": min(limit, 25),
                  "from": 0,
              }
          ]
      }
  
      if fields:
          payload["requests"][0]["fields"] = fields
  
      items_returned = 0
  
      while True:
          result = request("POST", "/search/query", account_id, json=payload)
  
          if not result or "value" not in result:
              break
  
          for response in result["value"]:
              if "hitsContainers" in response:
                  for container in response["hitsContainers"]:
                      if "hits" in container:
                          for hit in container["hits"]:
                              if limit and items_returned >= limit:
                                  return
                              yield hit["resource"]
                              items_returned += 1
  
          if "@odata.nextLink" in result:
              break
  
          has_more = False
          for response in result.get("value", []):
              for container in response.get("hitsContainers", []):
                  if container.get("moreResultsAvailable"):
                      has_more = True
                      break
  
          if not has_more:
              break
  
          payload["requests"][0]["from"] += payload["requests"][0]["size"]

• src/microsoft_mcp/server.py:3-15 (registration)

  The server entrypoint imports the FastMCP instance 'mcp' from tools.py (where all @mcp.tool decorators register the tools including unified_search) and calls mcp.run() to start the MCP server, making the tool available.

  from .tools import mcp
  
  
  def main() -> None:
      if not os.getenv("MICROSOFT_MCP_CLIENT_ID"):
          print(
              "Error: MICROSOFT_MCP_CLIENT_ID environment variable is required",
              file=sys.stderr,
          )
          sys.exit(1)
  
      mcp.run()