Microsoft MCP

Instructions

Implementation Reference

• src/microsoft_mcp/tools.py:840-860 (handler)

  The main execution function for the 'search_files' tool. Decorated with @mcp.tool for automatic registration in FastMCP. Searches OneDrive files using the graph.search_query helper and formats results into a list of dicts with id, name, type, size, modified, download_url.

  @mcp.tool
  def search_files(
      query: str,
      account_id: str,
      limit: int = 50,
  ) -> list[dict[str, Any]]:
      """Search for files in OneDrive using the modern search API."""
      items = list(graph.search_query(query, ["driveItem"], account_id, limit))
  
      return [
          {
              "id": item["id"],
              "name": item["name"],
              "type": "folder" if "folder" in item else "file",
              "size": item.get("size", 0),
              "modified": item.get("lastModifiedDateTime"),
              "download_url": item.get("@microsoft.graph.downloadUrl"),
          }
          for item in items
      ]

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

  Supporting utility function graph.search_query that performs the POST to /search/query Microsoft Graph API, handles pagination via 'from' parameter, and yields resource hits. Called by search_files with entity_types=['driveItem'].

  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/tools.py:8-8 (registration)

  Creation of the FastMCP server instance 'mcp' to which all tools including search_files are registered via @mcp.tool decorators.

  mcp = FastMCP("microsoft-mcp")