unified_search
Search across Microsoft 365 resources like messages, events, files, and sites using a single API query. Specify entity types or search all available data for comprehensive results.
Instructions
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.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| account_id | Yes | ||
| entity_types | No | ||
| limit | No | ||
| query | Yes |
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()