view_relationships

Instructions

Implementation Reference

• src/iac_memory_mcp_server/db/entities.py:221-281 (handler)

  Core handler function that executes the view_relationships tool logic: queries the database for the specified entity, its observations, and related entities, formats the results 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))]

• src/iac_memory_mcp_server/tools/entity.py:118-147 (handler)

  MCP wrapper handler for view_relationships tool that calls the core execute function with additional logging and MCP error handling.

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

• src/iac_memory_mcp_server/db/tools.py:335-340 (schema)

  JSON schema defining the input parameters for the view_relationships tool (requires entity_id string).

  "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 view_relationships handler in the entity_tool_handlers dictionary, which is later aggregated into the global tool handlers.

  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:24-28 (registration)

  Global aggregation of tool handlers including entity_tool_handlers (containing view_relationships) into tool_handlers dict used by MCP call_tool handler.

  tool_handlers = {
      **terraform_tool_handlers,
      **ansible_tool_handlers,
      **entity_tool_handlers,
  }