MCP Kafka Schema Reg

Instructions

Implementation Reference

• export_tools.py:42-117 (handler)

  The main handler function for the 'export_schema' tool. It handles parameters, gets the appropriate registry client, calls the common export function, adds metadata and links, and returns structured output or error.

  @structured_output("export_schema", fallback_on_error=True)
  def export_schema_tool(
      subject: str,
      registry_manager,
      registry_mode: str,
      version: str = "latest",
      context: Optional[str] = None,
      format: str = "json",
      registry: Optional[str] = None,
  ) -> Union[Dict[str, Any], str]:
      """
      Export a single schema in the specified format.
  
      Args:
          subject: The subject name
          version: The schema version (default: latest)
          context: Optional schema context
          format: Export format (json, avro_idl)
          registry: Optional registry name (ignored in single-registry mode)
  
      Returns:
          Exported schema 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_schema(client, subject, version, context, format)
  
          if isinstance(result, dict):
              # Add structured output metadata
              result["registry_mode"] = registry_mode
              result["mcp_protocol_version"] = "2025-06-18"
  
              # Ensure required fields for export schema
              if "subject" not in result:
                  result["subject"] = subject
              if "version" not in result:
                  result["version"] = version if version != "latest" else 1
              if "format" not in result:
                  result["format"] = format
  
              # Ensure content field is present - this is required by the schema
              if "content" not in result:
                  # Generate content from schema field
                  if "schema" in result:
                      if format == "json":
                          result["content"] = json.dumps(result["schema"], indent=2)
                      else:
                          result["content"] = str(result["schema"])
                  else:
                      result["content"] = ""
  
              # Add resource links for dictionary results
              registry_name = _get_registry_name_for_linking(registry_mode, client, registry)
              result = add_links_to_response(
                  result,
                  "schema",
                  registry_name,
                  subject=subject,
                  version=version,
                  context=context,
              )
  
          return result
      except Exception as e:
          return create_error_response(str(e), error_code="SCHEMA_EXPORT_FAILED", registry_mode=registry_mode)

• schema_definitions.py:498-513 (schema)

  JSON Schema definition for the output of the 'export_schema' tool, ensuring structured responses with required fields like subject, version, format, and content.

  EXPORT_SCHEMA_SCHEMA = {
      "type": "object",
      "properties": {
          "subject": {"type": "string", "description": "Subject name"},
          "version": {"type": "integer", "minimum": 1, "description": "Schema version"},
          "format": {
              "type": "string",
              "enum": ["json", "avro-idl"],
              "description": "Export format",
          },
          "content": {"type": "string", "description": "Exported schema content"},
          "metadata": {"type": "object", "description": "Schema metadata"},
          **METADATA_FIELDS,
      },
      "required": ["subject", "version", "format", "content"],
      "additionalProperties": True,

• schema_definitions.py:993-993 (schema)

  Mapping of 'export_schema' tool name to its output schema in the TOOL_OUTPUT_SCHEMAS dictionary, used for validation.

  "export_schema": EXPORT_SCHEMA_SCHEMA,

• schema_registry_common.py:1192-1212 (helper)

  Core helper function implementing the schema export logic, called by the tool handler. Fetches schema data and optionally converts to Avro IDL format.

  def export_schema(
      client: RegistryClient,
      subject: str,
      version: str = "latest",
      context: Optional[str] = None,
      format: str = "json",
  ) -> Union[Dict[str, Any], str]:
      """Export a single schema in the specified format."""
      try:
          schema_data = get_schema_with_metadata(client, subject, version, context)
          if "error" in schema_data:
              return schema_data
  
          if format == "avro_idl":
              schema_str = schema_data.get("schema", "")
              return format_schema_as_avro_idl(schema_str, subject)
          else:
              return schema_data
      except Exception as e:
          return {"error": str(e)}