create_context_interactive
Create and manage metadata contexts interactively for Kafka Schema Registry. Elicits organizational details when metadata is missing, ensuring accurate and complete context setup.
Instructions
Interactive context creation with elicitation for metadata.
When context metadata is not provided, this tool will elicit organizational information from the user.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| context | Yes | ||
| description | No | ||
| environment | No | ||
| owner | No | ||
| registry | No | ||
| tags | No |
Implementation Reference
- interactive_tools.py:437-518 (handler)The main handler function for the 'create_context_interactive' tool. It checks if metadata fields are missing, elicits them using the elicitation system if needed, and then calls the core 'create_context_tool' to perform the actual context creation, augmenting the result with elicited metadata.
async def create_context_interactive( context: str, registry: Optional[str] = None, # Additional metadata that could be elicited description: Optional[str] = None, owner: Optional[str] = None, environment: Optional[str] = None, tags: Optional[List[str]] = None, # Core tool dependencies injected create_context_tool=None, registry_manager=None, registry_mode=None, auth=None, headers=None, schema_registry_url=None, ) -> Dict[str, Any]: """ Interactive context creation with elicitation for metadata. When context metadata is not provided, this tool will elicit organizational information from the user. """ try: # Check if we should elicit metadata (any metadata field is None) needs_elicitation = any([description is None, owner is None, environment is None, tags is None]) elicited_metadata = {} if needs_elicitation: logger.info(f"Context creation for '{context}' could benefit from metadata") # Create elicitation request for context metadata elicitation_request = create_context_metadata_elicitation(context_name=context) # Store the request for processing await elicitation_manager.create_request(elicitation_request) # Attempt elicitation with fallback response = await elicit_with_fallback(elicitation_request) if response and response.complete: # Extract metadata from response elicited_metadata = { "description": response.values.get("description") or description, "owner": response.values.get("owner") or owner, "environment": response.values.get("environment") or environment, "tags": (response.values.get("tags", "").split(",") if response.values.get("tags") else tags), } # Filter out empty values elicited_metadata = {k: v for k, v in elicited_metadata.items() if v} logger.info(f"Collected context metadata from elicitation: {list(elicited_metadata.keys())}") # Now proceed with the actual context creation result = create_context_tool( context=context, registry_manager=registry_manager, registry_mode=registry_mode, registry=registry, auth=auth, headers=headers, schema_registry_url=schema_registry_url, ) # Add metadata to successful result if isinstance(result, dict) and "error" not in result: result["elicitation_used"] = needs_elicitation if elicited_metadata: result["metadata"] = elicited_metadata result["metadata_note"] = ( "Context created with elicited metadata. Store this information in your documentation system." ) return result except Exception as e: logger.error(f"Error in interactive context creation: {str(e)}") return create_error_response( f"Interactive context creation failed: {str(e)}", error_code="INTERACTIVE_CONTEXT_ERROR", )