MCP Kafka Schema Reg

Instructions

Implementation Reference

• core_registry_tools.py:1474-1545 (handler)

  Handler function for the 'list_contexts' tool. Queries the Schema Registry /contexts endpoint (or uses client.get_contexts() in multi-mode), formats the response with registry metadata, MCP version, and adds HATEOAS links using add_links_to_response.

  @structured_output("list_contexts", fallback_on_error=True)
  def list_contexts_tool(
      registry_manager,
      registry_mode: str,
      registry: Optional[str] = None,
      auth=None,
      headers=None,
      schema_registry_url: str = "",
  ) -> Dict[str, Any]:
      """
      List all available schema contexts.
  
      Args:
          registry: Optional registry name (ignored in single-registry mode)
  
      Returns:
          Dictionary containing context names with structured validation and resource links
      """
      try:
          if registry_mode == "single":
              # Single-registry mode: use secure session approach
              client = registry_manager.get_registry()
              if client is None:
                  return create_error_response(
                      "No default registry configured",
                      error_code="REGISTRY_NOT_FOUND",
                      registry_mode="single",
                  )
  
              response = client.session.get(f"{client.config.url}/contexts", auth=client.auth, headers=client.headers)
              response.raise_for_status()
              contexts_list = response.json()
  
              # Convert to enhanced response format
              result = {
                  "contexts": contexts_list,
                  "registry_mode": "single",
                  "mcp_protocol_version": "2025-06-18",
              }
  
              # Add resource links
              registry_name = _get_registry_name(registry_mode, registry)
              result = add_links_to_response(result, "contexts_list", registry_name)
  
              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",
                  )
  
              contexts_list = client.get_contexts()
  
              # Convert to enhanced response format
              result = {
                  "contexts": contexts_list,
                  "registry": client.config.name,
                  "registry_mode": "multi",
                  "mcp_protocol_version": "2025-06-18",
              }
  
              # Add resource links
              result = add_links_to_response(result, "contexts_list", client.config.name)
  
              return result
      except Exception as e:
          return create_error_response(str(e), error_code="CONTEXT_LIST_FAILED", registry_mode=registry_mode)

• schema_definitions.py:456-477 (schema)

  JSON Schema definition for the output of list_contexts tool, including required 'contexts' array, optional registry field, _links for HATEOAS, and shared METADATA_FIELDS.

  LIST_CONTEXTS_SCHEMA = {
      "type": "object",
      "properties": {
          "contexts": {
              "type": "array",
              "items": {"type": "string"},
              "description": "List of context names",
          },
          "registry": {
              "type": "string",
              "description": "Registry name (multi-registry mode)",
          },
          "_links": {
              "type": "object",
              "description": "Navigation links to related resources",
              "additionalProperties": True,
          },
          **METADATA_FIELDS,
      },
      "required": ["contexts"],
      "additionalProperties": True,
  }

• schema_definitions.py:970-970 (registration)

  Registration of the list_contexts output schema in the master TOOL_OUTPUT_SCHEMAS dictionary used for validation.

  "list_contexts": LIST_CONTEXTS_SCHEMA,