ServicesTemporal Service

Dynamic API Implementation

View as Markdown

Overview

The dynamic API endpoint generation system automatically creates FastAPI endpoints from workflow metadata. This creates a single source of truth for workflow definitions and provides excellent maintainability.

System Components

WorkflowMetadataMixin

The WorkflowMetadataMixin allows workflows to define their API metadata directly in the workflow class:

  • workflow_description: Human-readable description
  • workflow_name: Human-readable workflow name
  • workflow_input_class: Pydantic input model
  • workflow_api_enabled: Set to True to expose the workflow through the API
  • workflow_api_endpoint: API endpoint path
  • workflow_namespace: Workflow namespace

Dynamic API Generator

The dynamic API generator:

  • Creates routes for the workflow registry’s workflows that set workflow_api_enabled = True; the registry rejects an API-enabled workflow with incomplete metadata at startup
  • Generates FastAPI endpoint functions dynamically
  • Handles proper routing, documentation, and parameter validation
  • Provides consistent error handling and response formatting

Workflow Examples

All workflows use the metadata system, including:

  • BackupWorkflow: /ngc/backup - “Backup network device configuration to Config Store”
  • DeployWorkflow: /ngc/deploy - “Deploy intended configuration to network device with approval workflow”
  • VpcCreationWorkflow: /ngc/vpc_creation - “Create VPC with route distinguisher assignment and VRF provisioning”
  • VpcDeletionWorkflow: /ngc/vpc_deletion - “Delete VPC and associated VRFs with validation checks”
  • ReprovisionWorkflow: /ngc/reprovision - “Reprovision a network device using pre- and post-ZTP backups”
  • ConnectedHostMetadataWorkflow: /ngc/connected_host_metadata - “Discover and analyze connected hosts through MAC table and LLDP neighbor data”
  • HelloWorld: /hello_world - “Simple hello world workflow for testing and demonstration”
  • HelloWorldApproval: /hello_world_approval - “Hello world workflow with approval step for testing staged workflows”

Integrated Benefits

  • CLI automatically discovers workflows and generates commands
  • OpenAPI documentation is automatically generated
  • Consistent parameter validation across all interfaces
  • Single source of truth for workflow definitions

Technical Implementation

Workflow Definition with Metadata

# Workflows define their metadata directly in the class
@workflow.defn
class BackupWorkflow(WorkflowMetadataMixin, StageMixin, DeviceMixin, ArchiveMixin):
    """Network device configuration backup workflow."""
    # Workflow metadata
    workflow_name = "Backup"
    workflow_description = "Backup network device configuration to Config Store"
    workflow_input_class = BackupInput
    workflow_api_enabled = True
    workflow_api_endpoint = "/ngc/backup"
    workflow_namespace = "ngc"
    @workflow.run
    async def run(self, input_data: BackupInput) -> str:
        # Workflow implementation
        pass

Automatic Endpoint Generation

# In dynamic_endpoints.py - one POST route per API-enabled registry workflow
def register_dynamic_endpoints(router: APIRouter, workflows: Sequence[type] | None = None) -> None:
    """Register all workflow endpoints dynamically based on metadata."""
    for workflow_class in WORKFLOW_API_CATALOG if workflows is None else workflows:
        endpoint_path = workflow_class.get_workflow_api_endpoint()
        input_class = workflow_class.get_workflow_input_class()
        endpoint_func = create_workflow_endpoint(workflow_class, input_class, endpoint_path)
        router.post(
            endpoint_path,
            response_model=WorkflowResponse,
            summary=f"Execute {workflow_class.__name__}",
            description=workflow_class.get_workflow_description(),
        )(endpoint_func)

Key Benefits

Single Source of Truth

  • Workflow metadata is defined once in the workflow class
  • API endpoints, CLI help, and documentation all derive from the same source
  • Consistent information across all interfaces

Automatic Discovery

  • New workflows are automatically discovered and exposed using the API
  • CLI automatically generates commands for new workflows
  • No manual endpoint registration required

Easy Maintenance

  • Adding a new workflow requires only:
    1. Adding WorkflowMetadataMixin to the workflow class
    2. Setting the metadata attributes, including workflow_api_enabled = True
    3. Adding to BUILTIN_WORKFLOWS or to a workflow plugin descriptor
  • System automatically handles the rest

Comprehensive Documentation

  • Workflow descriptions appear in:
    • OpenAPI/Swagger documentation
    • CLI help text
    • Workflow discovery endpoints
  • Consistent documentation across all interfaces

Type Safety

  • Pydantic input models ensure type safety
  • FastAPI automatically generates request/response schemas
  • CLI parameter validation matches API validation

Architecture

Workflow Discovery Flow

Workflow Class WorkflowMetadataMixin Metadata Attributes Dynamic Endpoint Generator FastAPI Router CLI Command Generator OpenAPI Documentation

Components

WorkflowMetadataMixin (packages/workflows/src/nv_config_manager_workflows/metadata/mixin.py)

class WorkflowMetadataMixin:
    """Mixin for defining workflow metadata."""
    workflow_name: str | None = None
    workflow_description: str | None = None
    workflow_input_class: type[BaseModel] | None = None
    workflow_api_enabled: bool = False
    workflow_api_endpoint: str | None = None
    workflow_namespace: str | None = None
    @classmethod
    def get_workflow_description(cls) -> str:
        """Get the workflow description."""
        return cls.workflow_description or cls.__doc__.strip().split('\n')[0]
    @classmethod
    def has_complete_metadata(cls) -> bool:
        """Check if the workflow has complete API metadata defined."""
        return (
            cls.workflow_name is not None
            and cls.workflow_description is not None
            and cls.workflow_input_class is not None
            and cls.workflow_api_endpoint is not None
        )

Dynamic Endpoint Generator (src/nv_config_manager/temporal/api/dynamic_endpoints.py)

def create_workflow_endpoint(workflow_class: Type, input_class: Type[BaseModel], endpoint_path: str):
    """Create a FastAPI endpoint function for a workflow."""
    async def workflow_endpoint(body: BaseModel, request: Request) -> WorkflowResponse:
        """Execute the workflow with provided parameters."""
        if start_workflow is None:
            raise RuntimeError("start_workflow function not set.")
        # Auto-populate user fields from request auth data
        user = getattr(request.state, 'user', 'unknown')
        if hasattr(body, 'user') and not body.user:
            body.user = user
        if hasattr(body, 'user_domain') and not body.user_domain:
            body.user_domain = user.split("@")[1] if "@" in user else "nvidia.com"
        workflow_id = await start_workflow(request, workflow_class, body)
        return WorkflowResponse(id=workflow_id)
    # Set function metadata for FastAPI
    workflow_endpoint.__name__ = f"{workflow_class.__name__.lower()}_endpoint"
    workflow_endpoint.__doc__ = f"Execute the {workflow_class.__name__} workflow."
    workflow_endpoint.__annotations__ = {
        'body': input_class,
        'request': Request,
        'return': WorkflowResponse
    }
    return workflow_endpoint

Registration System

The API builds one validated workflow registry snapshot when it is imported. If the registry cannot be built, because an installed plugin is invalid or conflicts with another, or because the built-in plugin is not registered (WorkflowPluginDiscoveryError), the API fails to start. The API’s dynamic routes, /metadata, and /types read from the snapshot:

# In workflow_catalog.py
WORKFLOW_REGISTRY = build_workflow_registry()
# API-enabled workflows: dynamic POST routes and /metadata.
WORKFLOW_API_CATALOG = tuple(WORKFLOW_REGISTRY.api_workflows)
# Every registered workflow, including API-disabled child workflows: /types.
WORKFLOW_TYPE_CATALOG = tuple(WORKFLOW_REGISTRY.all_workflows)
# In dynamic_endpoints.py
def register_dynamic_endpoints(router: APIRouter, workflows: Sequence[type] | None = None) -> None:
    """Register all workflow endpoints dynamically based on metadata."""
    for workflow_class in WORKFLOW_API_CATALOG if workflows is None else workflows:
        endpoint_path = workflow_class.get_workflow_api_endpoint()
        input_class = workflow_class.get_workflow_input_class()
        endpoint_func = create_workflow_endpoint(workflow_class, input_class, endpoint_path)
        router.post(
            endpoint_path,
            response_model=WorkflowResponse,
            summary=f"Execute {workflow_class.__name__}",
            description=workflow_class.get_workflow_description(),
        )(endpoint_func)

Adding New Workflows

Create Workflow with Metadata

from nv_config_manager_workflows.metadata import WorkflowMetadataMixin
@workflow.defn
class MyNewWorkflow(WorkflowMetadataMixin, StageMixin):
    """My new workflow for doing amazing things."""
    # Workflow metadata
    workflow_name = "My New Workflow"
    workflow_description = "Perform amazing network operations with validation"
    workflow_input_class = MyNewWorkflowInput
    workflow_api_enabled = True
    workflow_api_endpoint = "/ngc/my_new_workflow"
    workflow_namespace = "ngc"
    @workflow.run
    async def run(self, input_data: MyNewWorkflowInput) -> str:
        # Workflow implementation
        pass

Create Input Model

class MyNewWorkflowInput(BaseModel):
    """Input model for MyNewWorkflow."""
    device_id: str
    operation_type: str
    validate_before: bool = True

Register Workflow

Add a built-in workflow to BUILTIN_WORKFLOWS:

# In packages/workflows/src/nv_config_manager_workflows/workflows/builtin.py
BUILTIN_WORKFLOWS = (
    # keep-sorted start
    # ... existing workflows
    MyNewWorkflow,
    # keep-sorted end
)

A workflow shipped in its own distribution belongs in that plugin’s WorkflowPluginDescriptor instead; see Registering a workflow plugin.

That’s It

The workflow is now automatically:

  • Available using the API at /v1/workflow/ngc/my_new_workflow
  • Documented in OpenAPI/Swagger
  • Available in CLI as uv run workflow-cli my-new-workflow
  • Included in workflow discovery endpoints

Workflow Discovery API

List Workflow Executions

GET /v1/workflow/

Returns workflow execution summaries. Query parameters such as workflow_type, status, user, site, device_name, device_id, device_role, and device_platform filter the execution list.

Get Registered Workflow Types

GET /v1/workflow/types

Returns the class name of every registered workflow, including API-disabled child workflows such as TenantDeployWorkflow. Use /v1/workflow/metadata for the set of workflows the API can start.

Get Workflow Metadata

GET /v1/workflow/metadata

Returns metadata for each API-enabled workflow and the RBAC roles needed by the UI:

{
  "workflows": [
    {
      "name": "BackupWorkflow",
      "display_name": "Backup",
      "description": "Backup network device configuration to Config Store",
      "endpoint": "/ngc/backup",
      "namespace": "ngc",
      "cli_name": "backup",
      "input_class": "BackupInput",
      "read_roles": ["all"],
      "execute_roles": ["all"]
    }
  ]
}

Testing

Unit Tests

def test_workflow_metadata():
    """Test that workflow has complete metadata."""
    assert BackupWorkflow.has_complete_metadata()
    assert BackupWorkflow.get_workflow_description() == "Backup network device configuration to Config Store"
    assert BackupWorkflow.get_workflow_api_endpoint() == "/ngc/backup"
    assert BackupWorkflow.get_workflow_input_class() == BackupInput
def test_dynamic_endpoint_generation():
    """Test that endpoints are generated correctly."""
    router = APIRouter()
    register_dynamic_endpoints(router)
    # Check that backup endpoint was registered
    backup_route = next((route for route in router.routes if route.path == "/ngc/backup"), None)
    assert backup_route is not None
    assert "POST" in backup_route.methods

Integration Tests

def test_backup_workflow_endpoint():
    """Test the dynamically generated backup endpoint."""
    response = client.post(
        "/v1/workflow/ngc/backup",
        json={"device_id": "test-device-id"},
        headers={"ssl-client-cert": test_cert}
    )
    assert response.status_code == 200
    assert "id" in response.json()
    assert "href" in response.json()

Summary

The dynamic API system provides a clean, maintainable approach to workflow management. By defining metadata directly in workflow classes, the system automatically generates consistent APIs, CLI commands, and documentation. This approach scales well as new workflows are added and ensures consistency across all interfaces.