IAC Memory MCP Server

Instructions

Implementation Reference

• src/iac_memory_mcp_server/tools/entity.py:27-58 (handler)

  MCP handler function for the create_entity tool that performs validation and delegates to execute_create_entity.

  async def handle_create_entity(db: Any, arguments: Dict[str, Any], operation_id: str) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
      """Handle create_entity tool."""
      # Validate required arguments
      if not arguments.get("name") or not arguments.get("type"):
          raise ValidationError("Missing required arguments: name and type are required")
  
      logger.info(
          "Creating entity",
          extra={
              "entity_type": arguments.get("type"),
              "operation_id": operation_id,
          },
      )
  
      try:
          # Execute creation
          return await execute_create_entity(db, arguments)
  
      except Exception as e:
          error_msg = f"Failed to create 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": "create_entity",
                      "operation_id": operation_id,
                  },
              )
          )

• src/iac_memory_mcp_server/db/tools.py:308-317 (schema)

  JSON schema definition for the create_entity tool parameters.

  "create_entity": {
      "type": "object",
      "description": "Create a new entity in the knowledge graph with optional initial observations",
      "required": ["name", "type"],
      "properties": {
          "name": {"type": "string", "description": "Entity name"},
          "type": {"type": "string", "description": "Entity type"},
          "observation": {"type": "string", "description": "Initial observation"},
      },
  },

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

  Registration of the create_entity handler in the entity_tool_handlers dictionary.

  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:143-184 (helper)

  Core execution logic for creating an entity in the database, including optional observation, called by the handler.

  async def execute_create_entity(
      db: DatabaseManager, arguments: Dict[str, Any]
  ) -> List[TextContent]:
      """Execute create entity operation.
  
      Args:
          db: Database manager instance
          arguments: Tool arguments
      """
      logger.info("Creating new entity", extra={"tool_arguments": arguments})
  
      with db.get_connection() as conn:
          conn.execute("PRAGMA busy_timeout = 5000")  # 5s timeout
          conn.execute("BEGIN IMMEDIATE")
          try:
              # Create entity
              cursor = conn.execute(
                  """INSERT INTO entities (name, type)
                  VALUES (?, ?)""",
                  (arguments["name"], arguments["type"]),
              )
              entity_id = cursor.lastrowid
  
              # Add observation if provided
              if "observation" in arguments:
                  conn.execute(
                      "INSERT INTO observations (entity_id, content) VALUES (?, ?)",
                      (entity_id, arguments["observation"]),
                  )
  
              conn.commit()
              return [
                  TextContent(
                      type="text",
                      text=f"Created entity '{arguments['name']}' (ID: {entity_id})",
                  )
              ]
          except Exception as e:
              conn.rollback()
              logger.error(f"Failed to create entity: {str(e)}")
              raise DatabaseError(f"Failed to create entity: {str(e)}")