MongoDB MCP Server

Instructions

Implementation Reference

• src/index.ts:460-549 (registration)

  Registration of the 'aggregate' tool in the ListTools response, including name, description, and input schema definition.

          {
            name: 'aggregate',
            description: `Execute a read-only aggregation pipeline on a collection.
  
  Supported Stages:
  - $match: Filter documents
  - $group: Group documents by a key
  - $sort: Sort documents
  - $project: Shape the output
  - $addFields - Include additional or calculated fields
  - $lookup: Perform left outer joins
  - $unwind: Deconstruct array fields
  
  
  Unsafe/Blocked Stages:
  - $out: Write results to collection
  - $merge: Merge results into collection
  - $set: Set field values
  - $unset: Remove fields
  - $replaceRoot: Replace document structure
  - $replaceWith: Replace document
  
  Example - User Statistics by Role:
  use_mcp_tool with
    server_name: "mongodb",
    tool_name: "aggregate",
    arguments: {
      "collection": "users",
      "pipeline": [
        { "$match": { "active": true } },
        { "$group": {
            "_id": "$role",
            "count": { "$sum": 1 },
            "avgAge": { "$avg": "$age" }
        }},
        { "$sort": { "count": -1 } }
      ],
      "limit": 100
    }
  
  Example - Posts with Author Details:
  use_mcp_tool with
    server_name: "mongodb",
    tool_name: "aggregate",
    arguments: {
      "collection": "posts",
      "pipeline": [
        { "$match": { "published": true } },
        { "$lookup": {
            "from": "users",
            "localField": "authorId",
            "foreignField": "_id",
            "as": "author"
        }},
        { "$unwind": "$author" },
        { "$project": {
            "title": 1,
            "authorName": "$author.name",
            "publishDate": 1
        }}
      ]
    }`,
            inputSchema: {
              type: 'object',
              properties: {
                database: {
                  type: 'string',
                  description: 'Database name (optional if default database is configured)',
                },
                collection: {
                  type: 'string',
                  description: 'Collection name',
                },
                pipeline: {
                  type: 'array',
                  description: 'MongoDB aggregation pipeline stages (read-only operations only)',
                  items: {
                    type: 'object',
                  },
                },
                limit: {
                  type: 'number',
                  description: 'Maximum number of documents to return (optional)',
                  minimum: 1,
                  maximum: 1000,
                },
              },
              required: ['collection', 'pipeline'],
            },
          },

• src/index.ts:522-548 (schema)

  Input schema definition for the 'aggregate' tool specifying parameters like database, collection, pipeline, and limit.

  inputSchema: {
    type: 'object',
    properties: {
      database: {
        type: 'string',
        description: 'Database name (optional if default database is configured)',
      },
      collection: {
        type: 'string',
        description: 'Collection name',
      },
      pipeline: {
        type: 'array',
        description: 'MongoDB aggregation pipeline stages (read-only operations only)',
        items: {
          type: 'object',
        },
      },
      limit: {
        type: 'number',
        description: 'Maximum number of documents to return (optional)',
        minimum: 1,
        maximum: 1000,
      },
    },
    required: ['collection', 'pipeline'],
  },

• src/index.ts:1154-1189 (handler)

  Handler for the 'aggregate' tool: validates pipeline safety, executes MongoDB aggregation on the specified collection, applies optional limit, fetches results, generates visualization hints, and returns formatted JSON output.

  case 'aggregate': {
    const { database, collection, pipeline, limit } = request.params.arguments as {
      database?: string;
      collection: string;
      pipeline: any[];
      limit?: number;
    };
    const dbName = database || this.defaultDatabase;
    if (!dbName) {
      throw new McpError(
        ErrorCode.InvalidRequest,
        'Database name is required when no default database is configured'
      );
    }
  
    this.validateAggregationPipeline(pipeline);
  
    const db = client.db(dbName);
    let aggregation = db.collection(collection).aggregate(pipeline);
  
    if (limit) {
      aggregation = aggregation.limit(limit);
    }
  
    const results = await aggregation.toArray();
    const vizHint = this.generateVisualizationHint(results);
  
    return {
      content: [
        {
          type: 'text',
          text: JSON.stringify(results, null, 2) + (vizHint ? `\n\nVisualization Hint:\n${vizHint}` : ''),
        },
      ],
    };
  }

• src/index.ts:161-187 (helper)

  Helper function to validate the aggregation pipeline, ensuring it only contains read-only stages and blocks potentially data-modifying operations.

  private validateAggregationPipeline(pipeline: any[]): void {
    if (!Array.isArray(pipeline)) {
      throw new McpError(
        ErrorCode.InvalidRequest,
        'Aggregation pipeline must be an array'
      );
    }
  
    if (pipeline.length === 0) {
      throw new McpError(
        ErrorCode.InvalidRequest,
        'Aggregation pipeline cannot be empty'
      );
    }
  
    const unsafeStages = ['$out', '$merge', '$set', '$unset', '$replaceRoot', '$replaceWith'];
    const unsafeStageFound = pipeline.find(stage => 
      Object.keys(stage).some(key => unsafeStages.includes(key))
    );
  
    if (unsafeStageFound) {
      throw new McpError(
        ErrorCode.InvalidRequest,
        'Pipeline contains unsafe stages that could modify data. Only read-only operations are allowed.'
      );
    }
  }