execute_mongodb_query
Run MongoDB native queries against a Metabase database by specifying database ID, collection, and query. Supports aggregation pipelines and optional parameters.
Instructions
Execute a MongoDB native query against a Metabase database.
Args: database_id: The ID of the MongoDB database to query. collection: The MongoDB collection name. query: The MongoDB query (aggregation pipeline array or query object). native_parameters: Optional parameters for the query.
Returns: Query execution results.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| database_id | Yes | ||
| collection | Yes | ||
| query | Yes | ||
| native_parameters | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- server.py:366-420 (handler)The actual implementation of the 'execute_mongodb_query' tool. It executes a MongoDB native query against a Metabase database by building a payload with the collection and query, converting the query to JSON if needed, then POSTing to /dataset endpoint.
@mcp.tool async def execute_mongodb_query( database_id: int, collection: str, query: Any, ctx: Context, native_parameters: list[dict[str, Any]] | None = None ) -> dict[str, Any]: """ Execute a MongoDB native query against a Metabase database. Args: database_id: The ID of the MongoDB database to query. collection: The MongoDB collection name. query: The MongoDB query (aggregation pipeline array or query object). native_parameters: Optional parameters for the query. Returns: Query execution results. """ try: import json await ctx.info(f"Executing MongoDB query on database {database_id}, collection {collection}") # Convert query to JSON string if it's not already a string if isinstance(query, (list, dict)): query_string = json.dumps(query) await ctx.debug(f"Converted query object to JSON string") else: query_string = str(query) payload = { "database": database_id, "type": "native", "native": { "query": query_string, "collection": collection } } if native_parameters: payload["native"]["parameters"] = native_parameters await ctx.debug(f"Query parameters: {len(native_parameters)} parameters provided") result = await metabase_client.request("POST", "/dataset", json=payload) row_count = len(result.get("data", {}).get("rows", [])) await ctx.info(f"MongoDB query executed successfully, returned {row_count} rows") return result except Exception as e: error_msg = f"Error executing MongoDB query: {e}" await ctx.error(error_msg) raise ToolError(error_msg) from e - server.py:366-367 (registration)The tool is registered via the @mcp.tool decorator on the execute_mongodb_query async function. This is the registration mechanism for the FastMCP framework.
@mcp.tool async def execute_mongodb_query(