MCP Think Tank

Instructions

Implementation Reference

• src/memory/tools.ts:76-157 (handler)

  The execute function implementing the core logic of the upsert_entities tool. It processes a list of entities, creating new ones by adding observations, updating existing ones if specified (replacing observations), with timeout protection and detailed results reporting.

  execute: async (args, context) => {
    const total = args.entities.length;
    const log = context && context.log ? context.log : { info() {}, error() {}, warn() {}, debug() {} };
    
    const results = {
      created: [] as string[],
      updated: [] as string[],
      failed: [] as {name: string, reason: string}[],
      incomplete: false
    };
  
    // Process with timeout protection
    const startTime = Date.now();
    const timeoutMs = 25000; // 25 second timeout
    
    for (const entity of args.entities) {
      try {
        // Check if we've exceeded our time budget
        if (Date.now() - startTime > timeoutMs) {
          results.incomplete = true;
          break;
        }
        
        // Check if entity exists
        const existingEntity = graph.entities.get(entity.name);
        const doesExist = !!existingEntity;
        
        // If entity doesn't exist, create it
        if (!doesExist) {
          // Add observations for the new entity
          for (const observation of entity.observations) {
            await memoryStore.add(entity.name, observation, {
              version: '1.0'
            });
          }
          results.created.push(entity.name);
        } 
        // If entity exists and update flag is true, update it
        else if (entity.update) {
          // Update entity type
          existingEntity.entityType = entity.entityType;
          
          // Remove all existing observations and add new ones
          graph.deleteObservations(entity.name, existingEntity.observations);
          
          // Add new observations
          for (const observation of entity.observations) {
            await memoryStore.add(entity.name, observation, {
              version: '1.0'
            });
          }
          results.updated.push(entity.name);
        }
        // Entity exists but update flag is false, skip
        else {
          results.failed.push({
            name: entity.name,
            reason: "Entity already exists and update flag is false"
          });
        }
      } catch (error) {
        results.failed.push({
          name: entity.name,
          reason: `Error processing entity: ${error}`
        });
      }
    }
    
    // Save final changes
    await memoryStore.save();
    
    // Return detailed results
    return JSON.stringify({
      created: results.created.length > 0 ? results.created : null,
      updated: results.updated.length > 0 ? results.updated : null,
      failed: results.failed.length > 0 ? results.failed : null,
      incomplete: results.incomplete,
      message: `Created ${results.created.length} new entities. Updated ${results.updated.length} existing entities. Failed for ${results.failed.length} entities.${
        results.incomplete ? ` Operation incomplete due to timeout - ${results.created.length + results.updated.length + results.failed.length} of ${total} entities processed.` : ''
      }`
    });
  }

• src/utils/validation.ts:19-27 (schema)

  Zod schema defining the input structure for upsert_entities: array of entities with name, entityType, observations, optional context and update flag.

  export const UpsertEntitiesSchema = z.object({
    entities: z.array(z.object({
      name: z.string().min(1).describe('Unique identifier for the entity'),
      entityType: z.string().min(1).describe('Type classification of the entity'),
      observations: z.array(z.string()).describe('Facts or observations about the entity'),
      context: z.string().optional().describe('Optional context or situation relevant to this entity (e.g., project, meeting, or scenario)'),
      update: z.boolean().optional().default(false).describe('If true, will fully replace an existing entity; if false, will only create if entity doesn\'t exist')
    })).describe('Array of entities to create or update')
  });

• src/memory/tools.ts:72-158 (registration)

  Tool registration within registerMemoryTools function using FastMCP server.addTool, specifying name, description, schema reference, and inline execute handler.

  server.addTool({
    name: 'upsert_entities',
    description: 'Create new entities or update existing ones in the knowledge graph using an upsert pattern',
    parameters: Schemas.UpsertEntitiesSchema,
    execute: async (args, context) => {
      const total = args.entities.length;
      const log = context && context.log ? context.log : { info() {}, error() {}, warn() {}, debug() {} };
      
      const results = {
        created: [] as string[],
        updated: [] as string[],
        failed: [] as {name: string, reason: string}[],
        incomplete: false
      };
  
      // Process with timeout protection
      const startTime = Date.now();
      const timeoutMs = 25000; // 25 second timeout
      
      for (const entity of args.entities) {
        try {
          // Check if we've exceeded our time budget
          if (Date.now() - startTime > timeoutMs) {
            results.incomplete = true;
            break;
          }
          
          // Check if entity exists
          const existingEntity = graph.entities.get(entity.name);
          const doesExist = !!existingEntity;
          
          // If entity doesn't exist, create it
          if (!doesExist) {
            // Add observations for the new entity
            for (const observation of entity.observations) {
              await memoryStore.add(entity.name, observation, {
                version: '1.0'
              });
            }
            results.created.push(entity.name);
          } 
          // If entity exists and update flag is true, update it
          else if (entity.update) {
            // Update entity type
            existingEntity.entityType = entity.entityType;
            
            // Remove all existing observations and add new ones
            graph.deleteObservations(entity.name, existingEntity.observations);
            
            // Add new observations
            for (const observation of entity.observations) {
              await memoryStore.add(entity.name, observation, {
                version: '1.0'
              });
            }
            results.updated.push(entity.name);
          }
          // Entity exists but update flag is false, skip
          else {
            results.failed.push({
              name: entity.name,
              reason: "Entity already exists and update flag is false"
            });
          }
        } catch (error) {
          results.failed.push({
            name: entity.name,
            reason: `Error processing entity: ${error}`
          });
        }
      }
      
      // Save final changes
      await memoryStore.save();
      
      // Return detailed results
      return JSON.stringify({
        created: results.created.length > 0 ? results.created : null,
        updated: results.updated.length > 0 ? results.updated : null,
        failed: results.failed.length > 0 ? results.failed : null,
        incomplete: results.incomplete,
        message: `Created ${results.created.length} new entities. Updated ${results.updated.length} existing entities. Failed for ${results.failed.length} entities.${
          results.incomplete ? ` Operation incomplete due to timeout - ${results.created.length + results.updated.length + results.failed.length} of ${total} entities processed.` : ''
        }`
      });
    }
  });