MCP Kafka Schema Reg

Instructions

Implementation Reference

• batch_operations.py:36-167 (handler)

  Primary MCP tool handler for 'clear_context_batch'. Decorated with @structured_output specifying the tool name. Validates inputs, resolves registry, creates async task using task_manager, schedules execution of _execute_clear_context_batch, returns task_id and metadata for monitoring.
  @structured_output("clear_context_batch", fallback_on_error=True) def clear_context_batch_tool( context: str, registry_manager, registry_mode: str, registry: Optional[str] = None, delete_context_after: bool = True, dry_run: bool = True, ) -> Dict[str, Any]: """Clear all subjects in a context using application-level batch operations. ⚠️ APPLICATION-LEVEL BATCHING: This performs application-level batching by making individual JSON-RPC requests for each operation. JSON-RPC batching has been disabled per MCP 2025-06-18 specification compliance. **MEDIUM-DURATION OPERATION** - Uses task queue pattern. This operation runs asynchronously and returns a task_id immediately. Use get_task_status(task_id) to monitor progress and get results. Performance Notes: - Uses parallel processing with ThreadPoolExecutor for efficiency - Individual requests maintain protocol compliance - Client-side request coordination replaces JSON-RPC batching Args: context: The context to clear registry: The registry to operate on (uses default if not specified) delete_context_after: Whether to delete the context after clearing subjects dry_run: If True, only simulate the operation without making changes Returns: Task information with task_id for monitoring progress with structured validation """ try: # Resolve registry name BEFORE creating task (critical for task lookup) if registry is None: # Get first available registry for single-registry compatibility available_registries = registry_manager.list_registries() if available_registries: registry = available_registries[0] # list_registries() returns list of strings else: return create_error_response( "No registries available", error_code="NO_REGISTRIES_AVAILABLE", registry_mode=registry_mode, ) # Validate registry exists registry_client = registry_manager.get_registry(registry) if not registry_client: return create_error_response( f"Registry '{registry}' not found", error_code="REGISTRY_NOT_FOUND", registry_mode=registry_mode, ) # Create async task with resolved registry name task = task_manager.create_task( TaskType.CLEANUP, metadata={ "operation": "clear_context_batch", "context": context, "registry": registry, # Now always a resolved string, never None "delete_context_after": delete_context_after, "dry_run": dry_run, "batching_type": "application_level", # For clarity "jsonrpc_batching": False, # Explicitly disabled }, ) # Start async execution try: # Check if there's a running event loop asyncio.get_running_loop() asyncio.create_task( task_manager.execute_task( task, _execute_clear_context_batch, context=context, registry=registry, registry_manager=registry_manager, delete_context_after=delete_context_after, dry_run=dry_run, ) ) except RuntimeError: # No running event loop, use thread pool to run the task def run_task(): asyncio.run( task_manager.execute_task( task, _execute_clear_context_batch, context=context, registry=registry, registry_manager=registry_manager, delete_context_after=delete_context_after, dry_run=dry_run, ) ) thread = threading.Thread(target=run_task) thread.start() # Return structured response result = { "message": "Context cleanup started as async task (application-level batching)", "task_id": task.id, "task": task.to_dict(), "operation_info": { "operation": "clear_context_batch", "expected_duration": "medium", "async_pattern": "task_queue", "batching_type": "application_level", "jsonrpc_batching_disabled": True, "guidance": "Long-running operation using individual requests. Returns task_id immediately. Use get_task_status() to monitor progress.", "registry_mode": registry_mode, "performance_note": "Uses parallel individual requests instead of JSON-RPC batching for MCP 2025-06-18 compliance", }, "operation": "clear_context_batch", "dry_run": dry_run, "total_items": 1, # One context "successful": 0, # Will be updated by task "failed": 0, # Will be updated by task "registry_mode": registry_mode, "mcp_protocol_version": "2025-06-18", } return result except Exception as e: return create_error_response(str(e), error_code="BATCH_OPERATION_FAILED", registry_mode=registry_mode)
• batch_operations.py:169-369 (handler)

  Core execution logic for clear_context_batch tool. Fetches subjects from context, performs parallel deletions using ThreadPoolExecutor, handles dry_run and viewonly modes, computes performance metrics, updates task progress, returns comprehensive results.
  def _execute_clear_context_batch( context: str, registry: str, registry_manager, delete_context_after: bool = True, dry_run: bool = True, ) -> Dict[str, Any]: """Execute the actual context cleanup logic using individual requests. Performance Implementation: - Uses ThreadPoolExecutor for parallel individual requests - Replaces previous JSON-RPC batching with application-level coordination - Maintains efficiency while ensuring MCP 2025-06-18 compliance """ start_time = time.time() subjects_found = 0 subjects_deleted = 0 context_deleted = False errors = [] try: # Get the current task ID from task manager for progress updates current_task = None for task in task_manager.list_tasks(status=TaskStatus.RUNNING): if ( task.metadata and task.metadata.get("operation") == "clear_context_batch" and task.metadata.get("context") == context and task.metadata.get("registry") == registry ): current_task = task break def update_progress(progress: float, message: str = ""): if current_task: task_manager.update_progress(current_task.id, progress) if message: logger.info(f"Clear Context Progress {progress:.1f}%: {message}") # Get registry client (registry is already resolved, never None here) registry_client = registry_manager.get_registry(registry) update_progress( 5.0, f"Starting cleanup of context '{context}' in registry '{registry}' (individual requests)", ) if not registry_client: return { "subjects_found": 0, "subjects_deleted": 0, "context_deleted": False, "dry_run": dry_run, "duration_seconds": time.time() - start_time, "success_rate": 0.0, "performance": 0.0, "message": f"Registry '{registry}' not found", "error": f"Registry '{registry}' not found", "registry": registry, "batching_method": "application_level", } update_progress(10.0, "Registry client connected") # Check viewonly mode viewonly_check = registry_manager.is_viewonly(registry) if viewonly_check: return { "subjects_found": 0, "subjects_deleted": 0, "context_deleted": False, "dry_run": dry_run, "duration_seconds": time.time() - start_time, "success_rate": 0.0, "performance": 0.0, "message": f"Registry '{registry}' is in VIEWONLY mode", "error": f"Registry '{registry}' is in VIEWONLY mode", "registry": registry, "batching_method": "application_level", } update_progress(20.0, "Fetching subjects from context") # Get all subjects in the context subjects = registry_client.get_subjects(context) if isinstance(subjects, dict) and "error" in subjects: subjects = [] subjects_found = len(subjects) if subjects_found == 0: update_progress(100.0, "Context is already empty") return { "subjects_found": 0, "subjects_deleted": 0, "context_deleted": False, "dry_run": dry_run, "duration_seconds": time.time() - start_time, "success_rate": 100.0, "performance": 0.0, "message": f"Context '{context}' is already empty", "registry": registry, "batching_method": "application_level", } update_progress( 30.0, f"Found {subjects_found} subjects to {'delete' if not dry_run else 'analyze'} (using individual requests)", ) if dry_run: update_progress( 100.0, f"DRY RUN: Would delete {subjects_found} subjects using individual requests", ) return { "subjects_found": subjects_found, "subjects_deleted": 0, "context_deleted": delete_context_after, "dry_run": True, "duration_seconds": time.time() - start_time, "success_rate": 100.0, "performance": 0.0, "message": f"DRY RUN: Would delete {subjects_found} subjects from context '{context}' using individual requests", "registry": registry, "batching_method": "application_level", } update_progress( 40.0, f"Starting deletion of {subjects_found} subjects using parallel individual requests", ) # Delete subjects in parallel using individual requests (replaces JSON-RPC batching) with ThreadPoolExecutor(max_workers=10) as executor: futures = [] for subject in subjects: future = executor.submit(_delete_subject_from_context, registry_client, subject, context) futures.append(future) total_futures = len(futures) for i, future in enumerate(as_completed(futures)): try: if future.result(): subjects_deleted += 1 except Exception as e: errors.append(str(e)) # Update progress for deletions (40% to 85%) deletion_progress = 40.0 + ((i + 1) / total_futures) * 45.0 update_progress( deletion_progress, f"Deleted {subjects_deleted} of {subjects_found} subjects (individual requests)", ) update_progress(90.0, "Computing cleanup results") # Calculate metrics duration = time.time() - start_time success_rate = (subjects_deleted / subjects_found * 100) if subjects_found > 0 else 100.0 performance = subjects_deleted / duration if duration > 0 else 0.0 # Delete context if requested (not supported by Schema Registry API) if delete_context_after and subjects_deleted == subjects_found: context_deleted = False # Context deletion not supported by API update_progress(95.0, "Context deletion not supported by API") update_progress( 100.0, f"Cleanup completed - deleted {subjects_deleted} subjects using individual requests", ) return { "subjects_found": subjects_found, "subjects_deleted": subjects_deleted, "context_deleted": context_deleted, "dry_run": False, "duration_seconds": duration, "success_rate": success_rate, "performance": performance, "message": f"Successfully cleared context '{context}' - deleted {subjects_deleted} subjects using individual requests", "errors": errors if errors else None, "registry": registry, "batching_method": "application_level", "compliance_note": "Uses individual requests per MCP 2025-06-18 specification (JSON-RPC batching disabled)", } except Exception as e: return { "subjects_found": subjects_found, "subjects_deleted": subjects_deleted, "context_deleted": False, "dry_run": dry_run, "duration_seconds": time.time() - start_time, "success_rate": 0.0, "performance": 0.0, "message": f"Batch cleanup failed: {str(e)}", "error": str(e), "registry": registry, "batching_method": "application_level", }
• schema_definitions.py:723-754 (schema)

  BATCH_OPERATION_SCHEMA: JSON Schema defining the structured output format for batch operations including clear_context_batch.
  # Batch operation response BATCH_OPERATION_SCHEMA = { "type": "object", "properties": { "operation": {"type": "string", "description": "Type of batch operation"}, "dry_run": {"type": "boolean", "description": "Whether this was a dry run"}, "total_items": { "type": "integer", "minimum": 0, "description": "Total items processed", }, "successful": { "type": "integer", "minimum": 0, "description": "Number of successful operations", }, "failed": { "type": "integer", "minimum": 0, "description": "Number of failed operations", }, "errors": { "type": "array", "items": {"type": "string"}, "description": "Error messages from failed operations", }, "details": {"type": "object", "description": "Detailed operation results"}, **METADATA_FIELDS, }, "required": ["operation", "dry_run", "total_items", "successful", "failed"], "additionalProperties": True, }
• schema_definitions.py:1019-1021 (schema)

  TOOL_OUTPUT_SCHEMAS mapping: Associates 'clear_context_batch' tool name with BATCH_OPERATION_SCHEMA for output validation.
  "clear_context_batch": BATCH_OPERATION_SCHEMA, "clear_multiple_contexts_batch": BATCH_OPERATION_SCHEMA, # Task Management
• task_management.py:154-157 (registration)

  OPERATION_METADATA registration: Defines expected duration (MEDIUM) and async pattern (TASK_QUEUE) for the clear_context_batch tool, used for client guidance.
  "clear_context_batch": { "duration": OperationDuration.MEDIUM, "pattern": AsyncPattern.TASK_QUEUE, },