Building a Production Workflow Deploy API with FastAPI

Deploy API Implementation - Production Workflow Deployment System

This post covers how we implemented the Deploy feature, which exposes workflows built in XGen as API endpoints that external services can call.

Deployment Concept

When a user "deploys" a workflow they've built on the canvas, a unique API endpoint is generated for that workflow. External services call this endpoint to trigger the workflow and receive the result.

Deployment Model

class Deployment(Base):
    __tablename__ = "deployments"

    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
    workflow_id = Column(String, ForeignKey("workflows.id"), nullable=False)
    user_id = Column(String, ForeignKey("users.id"), nullable=False)
    api_key = Column(String, unique=True, nullable=False)
    name = Column(String(255), nullable=False)
    description = Column(String(1000), default="")
    is_active = Column(Boolean, default=True)
    rate_limit = Column(Integer, default=100)  # requests per minute
    version = Column(Integer, default=1)
    input_schema = Column(JSON, default=dict)
    created_at = Column(DateTime, default=datetime.utcnow)

    workflow = relationship("Workflow")

Deployment Service

import secrets

class DeployService:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def deploy_workflow(
        self, workflow_id: str, user_id: str, config: dict
    ) -> Deployment:
        # Validate the workflow exists
        workflow = await self.get_workflow(workflow_id)
        if not workflow:
            raise AppException(404, "Workflow not found", "NOT_FOUND")

        # Validate the workflow graph (cycle detection)
        executor = WorkflowExecutor(workflow.nodes, workflow.edges)
        executor.get_execution_order()  # raises on cycle

        # Generate API key
        api_key = f"xgen_{secrets.token_urlsafe(32)}"

        deployment = Deployment(
            workflow_id=workflow_id,
            user_id=user_id,
            api_key=api_key,
            name=config.get("name", workflow.name),
            description=config.get("description", ""),
            input_schema=config.get("input_schema", {}),
        )
        self.db.add(deployment)
        await self.db.flush()

        # Save a workflow snapshot (preserves state at deploy time)
        await self._save_snapshot(deployment.id, workflow)

        return deployment

Deployed Workflow Execution API

router = APIRouter(prefix="/api/v1/execute", tags=["deploy"])

@router.post("/{deployment_id}")
async def execute_deployed_workflow(
    deployment_id: str,
    request: Request,
    x_api_key: str = Header(...),
):
    # Validate API key
    deployment = await deploy_service.verify_api_key(deployment_id, x_api_key)
    if not deployment:
        raise HTTPException(status_code=403, detail="Invalid API key")

    if not deployment.is_active:
        raise HTTPException(status_code=403, detail="Deployment is inactive")

    # Check rate limit
    await rate_limiter.check(deployment_id, deployment.rate_limit)

    # Execute workflow
    body = await request.json()
    result = await workflow_executor.execute(
        deployment.workflow_id,
        inputs=body.get("inputs", {}),
    )

    return {
        "deployment_id": deployment_id,
        "version": deployment.version,
        "result": result,
    }

Rate Limiting

class RateLimiter:
    def __init__(self, redis_client):
        self.redis = redis_client

    async def check(self, key: str, limit: int, window: int = 60):
        current = await self.redis.incr(f"rate:{key}")
        if current == 1:
            await self.redis.expire(f"rate:{key}", window)
        if current > limit:
            raise HTTPException(
                status_code=429,
                detail=f"Rate limit exceeded: {limit} requests/{window}s",
            )

With the deploy feature in place, AI workflows built in XGen can now be consumed by websites, mobile apps, Slack bots, and other external services. API key-based authentication and rate limiting provide the security and stability needed for production use.