IaC Memory MCP Server

Instructions

Implementation Reference

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

  The primary MCP handler function for the 'create_entity' tool. Validates inputs, logs the operation, calls the execution helper, and handles errors appropriately.

  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/entities.py:143-184 (helper)

  Core database execution logic for creating an entity and optional observation. Performs transaction-safe INSERT operations and returns a formatted success message.

  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)}")

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

  JSON schema defining the input parameters, requirements, and descriptions for the 'create_entity' tool.

  "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)

  Local registration mapping the 'create_entity' tool name to its handler function, which is later merged into the global 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/tools/__init__.py:43-55 (registration)

  Top-level MCP server registration of the call_tool and list_tools methods, which dispatch to handlers including 'create_entity' via the merged tool_handlers dictionary.

  def register_tools(server: Server) -> None:
      """Register all tool handlers with the server."""
  
      @server.call_tool()
      async def call_tool(
          name: str, arguments: Dict[str, Any], ctx: RequestContext | None = None
      ):
          return await handle_call_tool(name, arguments, ctx)
  
      @server.list_tools()
      async def list_tools(ctx: RequestContext = None):
          return await handle_list_tools(ctx)