feat: manifest runner service

Author: pedroslopez

airbyte_cdk/manifest_runner/__init__.py

@@ -0,0 +1,41 @@
+"""
+API Models for the Manifest Runner Service.
+
+This package contains all Pydantic models used for API requests and responses.
+"""
+
+from .dicts import ConnectorConfig, Manifest
+from .manifest import (
+    FullResolveRequest,
+    ManifestResponse,
+    ResolveRequest,
+    StreamTestReadRequest,
+)
+from .stream import (
+    AuxiliaryRequest,
+    HttpRequest,
+    HttpResponse,
+    LogMessage,
+    StreamRead,
+    StreamReadPages,
+    StreamReadSlices,
+)
+
+__all__ = [
+    # Typed Dicts
+    "ConnectorConfig",
+    "Manifest",
+    # Manifest request/response models
+    "FullResolveRequest",
+    "ManifestResponse",
+    "StreamTestReadRequest",
+    "ResolveRequest",
+    # Stream models
+    "AuxiliaryRequest",
+    "HttpRequest",
+    "HttpResponse",
+    "LogMessage",
+    "StreamRead",
+    "StreamReadPages",
+    "StreamReadSlices",
+]

airbyte_cdk/manifest_runner/api_models/__init__.py

@@ -0,0 +1,17 @@
+"""
+Common API models shared across different endpoints.
+"""
+
+from pydantic import BaseModel, ConfigDict
+
+
+class Manifest(BaseModel):
+    """Base manifest model. Allows client generation to replace with proper JsonNode types."""
+
+    model_config = ConfigDict(extra="allow")
+
+
+class ConnectorConfig(BaseModel):
+    """Base connector configuration model. Allows client generation to replace with proper JsonNode types."""
+
+    model_config = ConfigDict(extra="allow")

airbyte_cdk/manifest_runner/api_models/dicts.py

@@ -0,0 +1,45 @@
+"""
+Manifest-related API models.
+
+These models define the request and response structures for manifest operations
+like reading, resolving, and full resolution.
+"""
+
+from typing import Any, List, Optional
+
+from pydantic import BaseModel, Field
+
+from .dicts import ConnectorConfig, Manifest
+
+
+class StreamTestReadRequest(BaseModel):
+    """Request to test read from a specific stream."""
+
+    manifest: Manifest
+    config: ConnectorConfig
+    stream_name: str
+    state: List[Any] = []
+    custom_components_code: Optional[str] = None
+    record_limit: int = Field(default=100, ge=1, le=5000)
+    page_limit: int = Field(default=5, ge=1, le=20)
+    slice_limit: int = Field(default=5, ge=1, le=20)
+
+
+class ResolveRequest(BaseModel):
+    """Request to resolve a manifest."""
+
+    manifest: Manifest
+
+
+class ManifestResponse(BaseModel):
+    """Response containing a manifest."""
+
+    manifest: Manifest
+
+
+class FullResolveRequest(BaseModel):
+    """Request to fully resolve a manifest."""
+
+    manifest: Manifest
+    config: ConnectorConfig
+    stream_limit: int = Field(default=100, ge=1, le=100)

airbyte_cdk/manifest_runner/api_models/manifest.py

@@ -0,0 +1,76 @@
+"""
+Stream-related API models.
+
+These models define the structure for stream reading operations and responses.
+They accurately reflect the runtime types returned by the CDK, particularly
+fixing type mismatches like slice_descriptor being a string rather than an object.
+"""
+
+from typing import Any, Dict, List, Optional
+
+from pydantic import BaseModel
+
+
+class HttpRequest(BaseModel):
+    """HTTP request details."""
+
+    url: str
+    headers: Optional[Dict[str, Any]]
+    http_method: str
+    body: Optional[str] = None
+
+
+class HttpResponse(BaseModel):
+    """HTTP response details."""
+
+    status: int
+    body: Optional[str] = None
+    headers: Optional[Dict[str, Any]] = None
+
+
+class LogMessage(BaseModel):
+    """Log message from stream processing."""
+
+    message: str
+    level: str
+    internal_message: Optional[str] = None
+    stacktrace: Optional[str] = None
+
+
+class AuxiliaryRequest(BaseModel):
+    """Auxiliary HTTP request made during stream processing."""
+
+    title: str
+    type: str
+    description: str
+    request: HttpRequest
+    response: HttpResponse
+
+
+class StreamReadPages(BaseModel):
+    """Pages of data read from a stream slice."""
+
+    records: List[object]
+    request: Optional[HttpRequest] = None
+    response: Optional[HttpResponse] = None
+
+
+class StreamReadSlices(BaseModel):
+    """Slices of data read from a stream."""
+
+    pages: List[StreamReadPages]
+    slice_descriptor: Optional[str]  # This is actually a string at runtime, not Dict[str, Any]
+    state: Optional[List[Dict[str, Any]]] = None
+    auxiliary_requests: Optional[List[AuxiliaryRequest]] = None
+
+
+class StreamRead(BaseModel):
+    """Complete stream read response with properly typed fields."""
+
+    logs: List[LogMessage]
+    slices: List[StreamReadSlices]
+    test_read_limit_reached: bool
+    auxiliary_requests: List[AuxiliaryRequest]
+    inferred_schema: Optional[Dict[str, Any]]
+    inferred_datetime_formats: Optional[Dict[str, str]]
+    latest_config_update: Optional[Dict[str, Any]]

airbyte_cdk/manifest_runner/api_models/stream.py

@@ -0,0 +1,17 @@
+from fastapi import FastAPI
+
+from ..manifest_runner.routers import capabilities, health, manifest
+
+app = FastAPI(
+    title="Manifest Runner Service",
+    description="A service for running low-code Airbyte connectors",
+    version="0.1.0",
+    contact={
+        "name": "Airbyte",
+        "url": "https://airbyte.com",
+    },
+)
+
+app.include_router(health.router)
+app.include_router(capabilities.router)
+app.include_router(manifest.router, prefix="/v1")

airbyte_cdk/manifest_runner/app.py

@@ -0,0 +1,43 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+"""Main entry point for the Airbyte CDK Manifest Runner server."""
+
+import sys
+
+import uvicorn
+
+
+def check_dependencies() -> bool:
+    """Check if all required dependencies are available."""
+    try:
+        import fastapi
+        import uvicorn
+
+        return True
+    except ImportError:
+        return False
+
+
+def run_server(
+    host: str = "127.0.0.1", port: int = 8000, reload: bool = False, log_level: str = "info"
+) -> None:
+    """Run the FastAPI server."""
+    if not check_dependencies():
+        print("❌ Manifest runner dependencies not found. Please install with:")
+        print("  pip install airbyte-cdk[manifest-runner]")
+        print("  # or")
+        print("  poetry install --extras manifest-runner")
+        sys.exit(1)
+
+    print(f"🚀 Starting Airbyte CDK Manifest Runner on {host}:{port}")
+
+    uvicorn.run(
+        "airbyte_cdk.manifest_runner.app:app",
+        host=host,
+        port=port,
+        reload=reload,
+        log_level=log_level,
+    )
+
+
+if __name__ == "__main__":
+    run_server()

airbyte_cdk/manifest_runner/main.py

@@ -0,0 +1,46 @@
+from typing import Any, List, Mapping
+
+from airbyte_cdk.connector_builder.models import StreamRead
+from airbyte_cdk.connector_builder.test_reader import TestReader
+from airbyte_cdk.models.airbyte_protocol import (
+    AirbyteStateMessage,
+    ConfiguredAirbyteCatalog,
+)
+from airbyte_cdk.sources.declarative.manifest_declarative_source import (
+    ManifestDeclarativeSource,
+)
+
+
+class ManifestRunner:
+    _source: ManifestDeclarativeSource
+
+    def __init__(self, source: ManifestDeclarativeSource) -> None:
+        self._source = source
+
+    def test_read(
+        self,
+        config: Mapping[str, Any],
+        catalog: ConfiguredAirbyteCatalog,
+        state: List[AirbyteStateMessage],
+        record_limit: int,
+        page_limit: int,
+        slice_limit: int,
+    ) -> StreamRead:
+        """
+        Test the read method of the source.
+        """
+
+        test_read_handler = TestReader(
+            max_pages_per_slice=page_limit,
+            max_slices=slice_limit,
+            max_record_limit=record_limit,
+        )
+
+        stream_read = test_read_handler.run_test_read(
+            source=self._source,
+            config=config,
+            configured_catalog=catalog,
+            state=state,
+        )
+
+        return stream_read

airbyte_cdk/manifest_runner/manifest_runner/__init__.py

@@ -0,0 +1,73 @@
+from typing import Any, Mapping
+
+from airbyte_cdk.models.airbyte_protocol import (
+    AirbyteStream,
+    ConfiguredAirbyteCatalog,
+    ConfiguredAirbyteStream,
+    DestinationSyncMode,
+    SyncMode,
+)
+from airbyte_cdk.sources.declarative.manifest_declarative_source import (
+    ManifestDeclarativeSource,
+)
+from airbyte_cdk.sources.declarative.parsers.model_to_component_factory import (
+    ModelToComponentFactory,
+)
+
+SHOULD_NORMALIZE_KEY = "__should_normalize"
+SHOULD_MIGRATE_KEY = "__should_migrate"
+
+
+def build_catalog(stream_name: str) -> ConfiguredAirbyteCatalog:
+    return ConfiguredAirbyteCatalog(
+        streams=[
+            ConfiguredAirbyteStream(
+                stream=AirbyteStream(
+                    name=stream_name,
+                    json_schema={},
+                    supported_sync_modes=[SyncMode.full_refresh, SyncMode.incremental],
+                ),
+                sync_mode=SyncMode.incremental,
+                destination_sync_mode=DestinationSyncMode.overwrite,
+            )
+        ]
+    )
+
+
+def should_migrate_manifest(manifest: Mapping[str, Any]) -> bool:
+    """
+    Determines whether the manifest should be migrated,
+    based on the presence of the "__should_migrate" key.
+
+    This flag is set by the UI.
+    """
+    return manifest.get(SHOULD_MIGRATE_KEY, False)
+
+
+def should_normalize_manifest(manifest: Mapping[str, Any]) -> bool:
+    """
+    Determines whether the manifest should be normalized,
+    based on the presence of the "__should_normalize" key.
+
+    This flag is set by the UI.
+    """
+    return manifest.get(SHOULD_NORMALIZE_KEY, False)
+
+
+def build_source(
+    manifest: Mapping[str, Any], config: Mapping[str, Any]
+) -> ManifestDeclarativeSource:
+    return ManifestDeclarativeSource(
+        source_config=manifest,
+        config=config,
+        normalize_manifest=should_normalize_manifest(manifest),
+        migrate_manifest=should_migrate_manifest(manifest),
+        emit_connector_builder_messages=True,
+        component_factory=ModelToComponentFactory(
+            emit_connector_builder_messages=True,
+            limit_pages_fetched_per_slice=None,  # TODO
+            limit_slices_fetched=None,  # TODO
+            disable_retries=True,
+            disable_cache=True,
+        ),
+    )