IaC Memory MCP Server

Instructions

Implementation Reference

• src/iac_memory_mcp_server/tools/entity.py:60-87 (handler)

  The main MCP tool handler for 'update_entity' that logs the operation, delegates to execute_update_entity, and handles errors by raising McpError.

  async def handle_update_entity(db: Any, arguments: Dict[str, Any], operation_id: str) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
      """Handle update_entity tool."""
      try:
          logger.info(
              "Updating entity",
              extra={
                  "entity_id": arguments.get("entity_id"),
                  "operation_id": operation_id,
              },
          )
  
          # Execute update
          return execute_update_entity(db, arguments)
  
      except Exception as e:
          error_msg = f"Failed to update entity: {str(e)}"
          logger.error(error_msg, extra={"operation_id": operation_id})
          raise McpError(
              types.ErrorData(
                  code=types.INTERNAL_ERROR,
                  message=error_msg,
                  data={
                      "tool": "update_entity",
                      "operation_id": operation_id,
                  },
              )
          )

• src/iac_memory_mcp_server/db/tools.py:318-328 (schema)

  JSON schema definition for the 'update_entity' tool inputs, defining required 'id' and optional properties like name, type, observation.

  "update_entity": {
      "type": "object",
      "description": "Update an existing entity's properties and add new observations",
      "required": ["id"],
      "properties": {
          "id": {"type": "string", "description": "Entity ID"},
          "name": {"type": "string", "description": "New name"},
          "type": {"type": "string", "description": "New type"},
          "observation": {"type": "string", "description": "New observation"},
      },
  },

• src/iac_memory_mcp_server/tools/entity.py:150-155 (registration)

  Registration of the 'update_entity' handler in the entity_tool_handlers dictionary used for tool dispatching.

  entity_tool_handlers = {
      "create_entity": handle_create_entity,
      "update_entity": handle_update_entity,
      "delete_entity": handle_delete_entity,
      "view_relationships": handle_view_relationships,
  }

• src/iac_memory_mcp_server/db/entities.py:186-206 (helper)

  Helper function that orchestrates the entity update: prepares updates dict, calls core update_entity, adds observation if present, returns MCP TextContent response.

  def execute_update_entity(
      db: DatabaseManager, arguments: Dict[str, Any]
  ) -> List[TextContent]:
      """Execute update entity operation."""
      logger.info("Updating entity", extra={"args": arguments})
  
      updates = {k: v for k, v in arguments.items() if k != "id"}
      success = update_entity(db, arguments["id"], updates)
      if not success:
          raise DatabaseError(f"Entity not found: {arguments['id']}")
  
      # Add observation if provided
      if "observation" in arguments:
          with db.get_connection() as conn:
              conn.execute(
                  "INSERT INTO observations (entity_id, content) VALUES (?, ?)",
                  (arguments["id"], arguments["observation"]),
              )
  
      return [TextContent(type="text", text=f"Updated entity {arguments['id']}")]

• src/iac_memory_mcp_server/db/entities.py:51-62 (helper)

  Core database update function that dynamically builds and executes SQL UPDATE query on entities table.

  def update_entity(db: DatabaseManager, entity_id: str, updates: Dict) -> bool:
      """Update an existing entity."""
      set_clause = ", ".join(f"{k} = ?" for k in updates.keys())
      values = tuple(updates.values()) + (entity_id,)
      query = f"UPDATE entities SET {set_clause} WHERE id = ?"
      try:
          with db.get_connection() as conn:
              cursor = conn.execute(query, values)
              return cursor.rowcount > 0
      except sqlite3.Error as e:
          raise DatabaseError(f"Update failed: {str(e)}")