view_relationships
Retrieve all relationships and observations for a specific entity in the IAC Memory MCP Server to analyze and manage Infrastructure-as-Code data effectively.
Instructions
Retrieve all relationships and observations for a specific entity
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| entity_id | Yes | Entity ID |
Input Schema (JSON Schema)
{
"description": "Retrieve all relationships and observations for a specific entity",
"properties": {
"entity_id": {
"description": "Entity ID",
"type": "string"
}
},
"required": [
"entity_id"
],
"type": "object"
}
Implementation Reference
- MCP handler function for the view_relationships tool. Logs the operation, calls the execute_view_relationships helper, and handles exceptions by raising McpError.async def handle_view_relationships( db: Any, arguments: Dict[str, Any], operation_id: str ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]: """Handle view_relationships tool.""" try: logger.info( "Viewing entity relationships", extra={ "entity_id": arguments.get("entity_id"), "operation_id": operation_id, }, ) # Execute relationship view return execute_view_relationships(db, arguments) except Exception as e: error_msg = f"Failed to view relationships: {str(e)}" logger.error(error_msg, extra={"operation_id": operation_id}) raise McpError( types.ErrorData( code=types.INTERNAL_ERROR, message=error_msg, data={ "tool": "view_relationships", "operation_id": operation_id, }, ) )
- JSON schema definition for the view_relationships tool input parameters, requiring entity_id."view_relationships": { "type": "object", "description": "Retrieve all relationships and observations for a specific entity", "required": ["entity_id"], "properties": {"entity_id": {"type": "string", "description": "Entity ID"}}, },
- src/iac_memory_mcp_server/tools/entity.py:150-155 (registration)Registration of the view_relationships handler in the entity_tool_handlers dictionary used by the MCP server.entity_tool_handlers = { "create_entity": handle_create_entity, "update_entity": handle_update_entity, "delete_entity": handle_delete_entity, "view_relationships": handle_view_relationships, }
- Core helper function that executes the database query to fetch entity details, observations, and relationships, then formats them into TextContent.def execute_view_relationships( db: DatabaseManager, arguments: Dict[str, Any] ) -> List[TextContent]: """Execute view relationships operation.""" logger.info("Viewing relationships", extra={"relationship_args": arguments}) entity_id = arguments["entity_id"] with db.get_connection() as conn: cursor = conn.execute( """SELECT e.id, e.name, e.type, e.created_at, e.updated_at, o.content as observation, r.relationship_type, e2.id as related_id, e2.name as related_name, e2.type as related_type, e2.created_at as related_created_at, e2.updated_at as related_updated_at FROM entities e LEFT JOIN observations o ON e.id = o.entity_id LEFT JOIN entity_relationships r ON e.id = r.source_id LEFT JOIN entities e2 ON r.target_id = e2.id WHERE e.id = ?""", (entity_id,), ) entity = cursor.fetchone() if not entity: raise DatabaseError(f"Entity not found: {entity_id}") result = [ f"Entity {entity_id}:", f"Name: {entity['name']}", f"Type: {entity['type']}", f"Created: {entity['created_at']}", f"Updated: {entity['updated_at']}", ] if entity["observation"]: result.extend( [ "", # Empty line for readability f"Observation: {entity['observation']}", ] ) if entity["related_name"]: result.extend( [ "", # Empty line for readability "Related Entity:", f" ID: {entity['related_id']}", f" Name: {entity['related_name']}", f" Type: {entity['related_type']}", f" Created: {entity['related_created_at']}", f" Updated: {entity['related_updated_at']}", f" Relationship Type: {entity['relationship_type']}", ] ) return [TextContent(type="text", text="\n".join(result))]