create_entity
Add new Infrastructure-as-Code components to persistent memory storage for version tracking and relationship mapping in Terraform and Ansible environments.
Instructions
Create a new entity in the knowledge graph with optional initial observations
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| name | Yes | Entity name | |
| type | Yes | Entity type | |
| observation | No | Initial observation |
Implementation Reference
- 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, }, ) )
- 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)}")
- 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)