Microsoft MCP

Instructions

Implementation Reference

• src/microsoft_mcp/tools.py:888-920 (handler)

  The handler function implementing the search_events tool logic, including registration via @mcp.tool decorator. Searches events via graph.search_query and applies optional date filtering.

  @mcp.tool
  def search_events(
      query: str,
      account_id: str,
      days_ahead: int = 365,
      days_back: int = 365,
      limit: int = 50,
  ) -> list[dict[str, Any]]:
      """Search calendar events using the modern search API."""
      events = list(graph.search_query(query, ["event"], account_id, limit))
  
      # Filter by date range if needed
      if days_ahead != 365 or days_back != 365:
          now = dt.datetime.now(dt.timezone.utc)
          start = now - dt.timedelta(days=days_back)
          end = now + dt.timedelta(days=days_ahead)
  
          filtered_events = []
          for event in events:
              event_start = dt.datetime.fromisoformat(
                  event.get("start", {}).get("dateTime", "").replace("Z", "+00:00")
              )
              event_end = dt.datetime.fromisoformat(
                  event.get("end", {}).get("dateTime", "").replace("Z", "+00:00")
              )
  
              if event_start <= end and event_end >= start:
                  filtered_events.append(event)
  
          return filtered_events
  
      return events

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

  Supporting utility function that performs the actual search query to Microsoft Graph /search/query endpoint, handling pagination and yielding results. Called by search_events.

  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"]