feat(cli): Enable creating Workflows from the CLI

tracecat/api/app.py

@@ -1,5 +1,6 @@
 import asyncio
 from contextlib import asynccontextmanager
+from datetime import datetime
 from pathlib import Path
 from typing import Annotated, Any
 
@@ -351,17 +352,30 @@ def list_workflows(
 
 @app.post("/workflows", status_code=status.HTTP_201_CREATED)
 def create_workflow(
-    role: Annotated[Role, Depends(authenticate_user)],
+    role: Annotated[Role, Depends(authenticate_user_or_service)],
     params: CreateWorkflowParams,
 ) -> WorkflowMetadataResponse:
     """Create new Workflow with title and description."""
-    # When we create a workflow, we automatically create a webhook
+
+    title = (
+        params.title
+        if params.title != Undefined.Value
+        else datetime.now().strftime("%b %d, %Y, %H:%M:%S")
+    )
+    # Create the message
+    description = (
+        params.description
+        if params.description != Undefined.Value
+        else f"New workflow created {title}"
+    )
+
     with Session(engine) as session:
         workflow = Workflow(
-            title=params.title,
-            description=params.description,
+            title=title,
+            description=description,
             owner_id=role.user_id,
         )
+        # When we create a workflow, we automatically create a webhook
         webhook = Webhook(
             owner_id=role.user_id,
             workflow_id=workflow.id,

tracecat/cli/config.py

@@ -1,6 +1,10 @@
+from dotenv import find_dotenv, load_dotenv
+
 from tracecat.auth import Role
 
+load_dotenv(find_dotenv())
+
+# In reality we should use the user's id from config.toml
 USER = "default-tracecat-user"
-SECRET = "test-secret"
 
 ROLE: Role = Role(type="service", user_id=USER, service_id="tracecat-runner")

tracecat/cli/workflow.py

@@ -1,19 +1,22 @@
 import asyncio
 from pathlib import Path
 
+import httpx
 import orjson
 import rich
 import typer
 
 from tracecat.auth import AuthenticatedAPIClient
 from tracecat.dsl.workflow import DSLInput
+from tracecat.types.api import WebhookResponse
 
 from . import config
 
 app = typer.Typer(no_args_is_help=True)
 
 
-async def upsert_workflow_definition(yaml_path: Path, workflow_id: str):
+async def _upsert_workflow_definition(yaml_path: Path, workflow_id: str):
+    """Bypass /workflows/{workflow_id}/commit endpoint and directly upsert the definition."""
     defn_content = DSLInput.from_yaml(yaml_path)
 
     async with AuthenticatedAPIClient(role=config.ROLE) as client:
@@ -22,12 +25,43 @@ async def upsert_workflow_definition(yaml_path: Path, workflow_id: str):
         res.raise_for_status()
 
 
-async def run_workflow(workflow_id: str, content: dict[str, str] | None = None):
+async def _run_workflow(workflow_id: str, content: dict[str, str] | None = None):
     async with AuthenticatedAPIClient(role=config.ROLE) as client:
-        res = await client.post(
-            f"/webhooks/{workflow_id}/{config.SECRET}", content=content
-        )
+        # Get the webhook url
+        res = await client.get(f"/workflows/{workflow_id}/webhooks")
         res.raise_for_status()
+        # There's only 1 webhook
+        webhooks = WebhookResponse.model_validate(res.json()[0])
+    async with httpx.AsyncClient() as client:
+        res = await client.post(webhooks.url, content=content)
+        res.raise_for_status()
+        rich.print(res.json())
+
+
+async def _create_workflow(title: str | None = None, description: str | None = None):
+    async with AuthenticatedAPIClient(role=config.ROLE) as client:
+        # Get the webhook url
+        params = {}
+        if title:
+            params["title"] = title
+        if description:
+            params["description"] = description
+        res = await client.post("/workflows", content=orjson.dumps(params))
+        res.raise_for_status()
+    rich.print("Created workflow")
+    rich.print(res.json())
+
+
+@app.command(help="Create a workflow")
+def create(
+    file: Path = typer.Option(
+        ..., "--file", "-f", help="Path to the workflow definition YAML file"
+    ),
+):
+    """Create a new workflow."""
+    rich.print("Creating a new workflow")
+    defn = DSLInput.from_yaml(file)
+    asyncio.run(_create_workflow(defn.title, defn.description))
 
 
 @app.command(help="List all workflow definitions")
@@ -40,7 +74,7 @@ def commit(
     ),
 ):
     """Commit a workflow definition to the database."""
-    asyncio.run(upsert_workflow_definition(file, workflow_id))
+    asyncio.run(_upsert_workflow_definition(file, workflow_id))
     rich.print(f"Upserted workflow definition for {workflow_id!r}")
 
 
@@ -59,4 +93,4 @@ def run(
 ):
     """Triggers a webhook to run a workflow."""
     rich.print(f"Running workflow {workflow_id!r}")
-    asyncio.run(run_workflow(workflow_id, orjson.loads(data) if data else None))
+    asyncio.run(_run_workflow(workflow_id, orjson.loads(data) if data else None))

tracecat/types/api.py

@@ -114,8 +114,8 @@ class WorkflowRunEventParams(BaseModel):
 
 
 class CreateWorkflowParams(BaseModel):
-    title: str
-    description: str
+    title: str | Undefined = Undefined.Value
+    description: str | Undefined = Undefined.Value
 
 
 class UpdateWorkflowParams(BaseModel):