MCP Kafka Schema Reg

Instructions

Implementation Reference

• export_tools.py:120-177 (handler)

  MCP tool handler function for 'export_subject'. Decorated with @structured_output, handles registry selection, calls core export function, adds MCP metadata and resource links.

  @structured_output("export_subject", fallback_on_error=True)
  def export_subject_tool(
      subject: str,
      registry_manager,
      registry_mode: str,
      context: Optional[str] = None,
      include_metadata: bool = True,
      include_config: bool = True,
      include_versions: str = "all",
      registry: Optional[str] = None,
  ) -> Dict[str, Any]:
      """
      Export all versions of a subject.
  
      Args:
          subject: The subject name
          context: Optional schema context
          include_metadata: Include export metadata
          include_config: Include subject configuration
          include_versions: Which versions to include (all, latest)
          registry: Optional registry name (ignored in single-registry mode)
  
      Returns:
          Dictionary containing subject export data with structured validation and resource links
      """
      try:
          if registry_mode == "single":
              client = get_default_client(registry_manager)
          else:
              client = registry_manager.get_registry(registry)
  
          if client is None:
              return create_error_response(
                  "No registry configured or registry not found",
                  error_code="REGISTRY_NOT_CONFIGURED",
                  registry_mode=registry_mode,
              )
  
          result = common_export_subject(client, subject, context, include_metadata, include_config, include_versions)
  
          # Add structured output metadata
          result["registry_mode"] = registry_mode
          result["mcp_protocol_version"] = "2025-06-18"
  
          # Ensure required fields for export subject
          if "subject" not in result:
              result["subject"] = subject
          if "versions" not in result:
              result["versions"] = []
  
          # Add resource links
          registry_name = _get_registry_name_for_linking(registry_mode, client, registry)
          result = add_links_to_response(result, "subject", registry_name, subject=subject, context=context)
  
          return result
      except Exception as e:
          return create_error_response(str(e), error_code="SUBJECT_EXPORT_FAILED", registry_mode=registry_mode)

• schema_definitions.py:517-553 (schema)

  JSON Schema definition for the output of export_subject tool, defining structure for subject versions, config, and metadata.

  EXPORT_SUBJECT_SCHEMA = {
      "type": "object",
      "properties": {
          "subject": {"type": "string", "description": "Subject name"},
          "versions": {
              "type": "array",
              "items": {
                  "type": "object",
                  "properties": {
                      "version": {"type": "integer", "minimum": 1},
                      "id": {"type": "integer", "minimum": 0},
                      "schema": {
                          "oneOf": [
                              {"type": "string", "description": "The schema definition as JSON string"},
                              {"type": "object", "description": "The schema definition as JSON object"},
                          ]
                      },
                      "schemaType": {"type": "string"},
                  },
                  "required": ["version", "id", "schema"],
              },
              "description": "All versions of the subject",
          },
          "config": {"type": "object", "description": "Subject configuration"},
          "export_metadata": {
              "type": "object",
              "properties": {
                  "exported_at": {"type": "string", "format": "date-time"},
                  "total_versions": {"type": "integer", "minimum": 0},
                  "include_config": {"type": "boolean"},
              },
          },
          **METADATA_FIELDS,
      },
      "required": ["subject", "versions"],
      "additionalProperties": True,
  }

• schema_definitions.py:993-994 (registration)

  Registration of export_subject output schema in the central TOOL_OUTPUT_SCHEMAS dictionary used for structured output validation.

  "export_schema": EXPORT_SCHEMA_SCHEMA,
  "export_subject": EXPORT_SUBJECT_SCHEMA,

• schema_registry_common.py:1215-1258 (helper)

  Core helper function implementing the subject export logic: fetches versions, retrieves schemas, includes config and metadata.

      client: RegistryClient,
      subject: str,
      context: Optional[str] = None,
      include_metadata: bool = True,
      include_config: bool = True,
      include_versions: str = "all",
  ) -> Dict[str, Any]:
      """Export all versions of a subject."""
      try:
          # Get versions
          if include_versions == "latest":
              versions = ["latest"]
          else:
              versions_list = client.get_schema_versions(subject, context)
              if isinstance(versions_list, dict) and "error" in versions_list:
                  return versions_list
              versions = [str(v) for v in versions_list]
  
          # Get schemas for each version
          schemas = []
          for version in versions:
              schema_data = get_schema_with_metadata(client, subject, version, context)
              if "error" not in schema_data:
                  schemas.append(schema_data)
  
          result = {"subject": subject, "versions": schemas}
  
          if include_config:
              config = client.get_subject_config(subject, context)
              if "error" not in config:
                  result["config"] = config
  
          if include_metadata:
              result["metadata"] = {
                  "exported_at": datetime.now().isoformat(),
                  "registry_url": client.config.url,
                  "context": context,
                  "export_version": "1.7.0",
              }
  
          return result
      except Exception as e:
          return {"error": str(e)}