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)}