MCP Kafka Schema Reg

Instructions

Implementation Reference

• export_tools.py:179-358 (handler)

  Primary MCP tool handler implementing export_context with progress reporting, registry mode handling (single/multi), error handling, resource linking, and delegation to core common_export_context function.

  @structured_output("export_context", fallback_on_error=True)
  async def export_context_tool(
      context: str,
      registry_manager,
      registry_mode: str,
      registry: Optional[str] = None,
      include_metadata: bool = True,
      include_config: bool = True,
      include_versions: str = "all",
      mcp_context: Optional["Context"] = None,
  ) -> Dict[str, Any]:
      """
      Export all subjects within a context.
  
      Args:
          context: The context name
          registry: Optional registry name (ignored in single-registry mode)
          include_metadata: Include export metadata
          include_config: Include configuration data
          include_versions: Which versions to include (all, latest)
          mcp_context: MCP Context for progress reporting
  
      Returns:
          Dictionary containing context export data with structured validation and resource links
      """
      try:
          # Initial setup (0-10%)
          if mcp_context:
              await mcp_context.info(f"Starting context export: {context}")
              await mcp_context.report_progress(0.0, 100.0, "Initializing context export")
  
          if registry_mode == "single":
              # Single-registry mode: use common function
              client = get_default_client(registry_manager)
              if client is None:
                  return create_error_response(
                      "No default registry configured",
                      error_code="REGISTRY_NOT_CONFIGURED",
                      registry_mode="single",
                  )
  
              if mcp_context:
                  await mcp_context.report_progress(5.0, 100.0, "Using default registry client")
  
              result = common_export_context(client, context, include_metadata, include_config, include_versions)
              result["registry_mode"] = "single"
              result["mcp_protocol_version"] = "2025-06-18"
  
              # Add resource links
              registry_name = _get_registry_name_for_linking(registry_mode, client, registry)
              result = add_links_to_response(result, "context", registry_name, context=context)
  
              if mcp_context:
                  await mcp_context.info("Context export completed successfully")
                  await mcp_context.report_progress(100.0, 100.0, "Context export completed")
  
              return result
          else:
              # Multi-registry mode: use client approach
              client = registry_manager.get_registry(registry)
              if client is None:
                  return create_error_response(
                      f"Registry '{registry}' not found",
                      error_code="REGISTRY_NOT_FOUND",
                      registry_mode="multi",
                  )
  
              if mcp_context:
                  await mcp_context.report_progress(10.0, 100.0, f"Registry client '{registry}' initialized")
  
              # Get all subjects in context (10-25%)
              if mcp_context:
                  await mcp_context.info(f"Fetching subjects from context: {context}")
                  await mcp_context.report_progress(15.0, 100.0, f"Fetching subjects from context '{context}'")
  
              subjects_list = client.get_subjects(context)
              if isinstance(subjects_list, dict) and "error" in subjects_list:
                  return create_error_response(
                      f"Failed to get subjects for context '{context}': {subjects_list.get('error')}",
                      error_code="CONTEXT_SUBJECTS_RETRIEVAL_FAILED",
                      registry_mode=registry_mode,
                  )
  
              if mcp_context:
                  await mcp_context.report_progress(25.0, 100.0, f"Found {len(subjects_list)} subjects in context")
  
              # Export each subject (25-70%)
              subjects_data = []
              if mcp_context:
                  await mcp_context.info(f"Exporting {len(subjects_list)} subjects")
                  await mcp_context.report_progress(30.0, 100.0, f"Starting export of {len(subjects_list)} subjects")
  
              for i, subject in enumerate(subjects_list):
                  # Report progress for subject export
                  if mcp_context and len(subjects_list) > 0:
                      progress = 30.0 + (i / len(subjects_list)) * 40.0  # 30% to 70%
                      await mcp_context.report_progress(
                          progress, 100.0, f"Exporting subject {i+1}/{len(subjects_list)}: {subject}"
                      )
  
                  subject_export = export_subject_tool(
                      subject,
                      registry_manager,
                      registry_mode,
                      context,
                      include_metadata,
                      include_config,
                      include_versions,
                      registry,
                  )
                  if "error" not in subject_export:
                      subjects_data.append(subject_export)
  
              if mcp_context:
                  await mcp_context.report_progress(70.0, 100.0, f"Exported {len(subjects_data)} subjects successfully")
  
              # Build result structure (70-80%)
              if mcp_context:
                  await mcp_context.report_progress(75.0, 100.0, "Building export result structure")
  
              result = {
                  "context": context,
                  "subjects": subjects_data,
                  "subject_count": len(subjects_data),
                  "registry": client.config.name,
                  "registry_mode": registry_mode,
                  "mcp_protocol_version": "2025-06-18",
              }
  
              # Add configuration data if requested (80-90%)
              if include_config:
                  if mcp_context:
                      await mcp_context.report_progress(80.0, 100.0, "Fetching context configuration")
  
                  global_config = client.get_global_config(context)
                  if "error" not in global_config:
                      result["global_config"] = global_config
  
                  global_mode = client.get_mode(context)
                  if "error" not in global_mode:
                      result["global_mode"] = global_mode
  
                  if mcp_context:
                      await mcp_context.report_progress(85.0, 100.0, "Context configuration added")
              else:
                  if mcp_context:
                      await mcp_context.report_progress(85.0, 100.0, "Skipping context configuration")
  
              # Add metadata if requested (90-95%)
              if include_metadata:
                  if mcp_context:
                      await mcp_context.report_progress(90.0, 100.0, "Adding export metadata")
  
                  from datetime import datetime
  
                  result["metadata"] = {
                      "exported_at": datetime.now().isoformat(),
                      "registry_url": client.config.url,
                      "registry_name": client.config.name,
                      "export_version": "2.0.0",
                      "registry_mode": "multi",
                  }
  
              # Add resource links (95-100%)
              if mcp_context:
                  await mcp_context.report_progress(95.0, 100.0, "Adding resource links")
  
              registry_name = _get_registry_name_for_linking(registry_mode, client, registry)
              result = add_links_to_response(result, "context", registry_name, context=context)
  
              if mcp_context:
                  await mcp_context.info("Context export completed successfully")
                  await mcp_context.report_progress(100.0, 100.0, "Context export completed")
  
              return result
      except Exception as e:
          if mcp_context:
              await mcp_context.error(f"Context export failed: {str(e)}")
          return create_error_response(str(e), error_code="CONTEXT_EXPORT_FAILED", registry_mode=registry_mode)

• schema_registry_common.py:1260-1307 (helper)

  Core helper function containing the fundamental export logic: lists subjects in the given context, recursively exports each subject's versions/config, assembles context-level result with optional global config, mode, and metadata.

  def export_context(
      client: RegistryClient,
      context: str,
      include_metadata: bool = True,
      include_config: bool = True,
      include_versions: str = "all",
  ) -> Dict[str, Any]:
      """Export all subjects within a context."""
      try:
          # Get all subjects in context
          subjects_list = client.get_subjects(context)
  
          # Export each subject
          subjects_data = []
          for subject in subjects_list:
              subject_export = export_subject(
                  client,
                  subject,
                  context,
                  include_metadata,
                  include_config,
                  include_versions,
              )
              if "error" not in subject_export:
                  subjects_data.append(subject_export)
  
          result = {"context": context, "subjects": subjects_data}
  
          if include_config:
              global_config = client.get_global_config(context)
              if "error" not in global_config:
                  result["global_config"] = global_config
  
              global_mode = client.get_mode(context)
              if "error" not in global_mode:
                  result["global_mode"] = global_mode
  
          if include_metadata:
              result["metadata"] = {
                  "exported_at": datetime.now().isoformat(),
                  "registry_url": client.config.url,
                  "export_version": "1.7.0",
              }
  
          return result
      except Exception as e:
          return {"error": str(e)}

• schema_definitions.py:995-998 (schema)

  JSON Schema output validation definition for the export_context tool (noted as complex structure allowing additional properties).

  "export_context": {
      "type": "object",
      "additionalProperties": True,
  },  # Complex export structure