Merge pull request #124 from askui/feat/create-thread-and-run-endpoint

feat/create thread and run endpoint

Author: adi-wan-askui

src/askui/chat/api/runs/models.py

@@ -1,9 +1,10 @@
 from datetime import datetime, timedelta, timezone
 from typing import Literal
 
-from pydantic import BaseModel, Field, computed_field
+from pydantic import BaseModel, computed_field
 
 from askui.chat.api.models import AssistantId, RunId, ThreadId
+from askui.chat.api.threads.models import ThreadCreateParams
 from askui.utils.api_utils import Resource
 from askui.utils.datetime_utils import UnixDatetime, now
 from askui.utils.id_utils import generate_time_ordered_id
@@ -38,6 +39,10 @@ class RunCreateParams(RunBase):
     stream: bool = False
 
 
+class ThreadAndRunCreateParams(RunCreateParams):
+    thread: ThreadCreateParams
+
+
 class Run(RunBase, Resource):
     """A run execution within a thread."""
 

src/askui/chat/api/runs/router.py

@@ -13,13 +13,13 @@
 from askui.utils.api_utils import ListQuery, ListResponse
 
 from .dependencies import RunServiceDep
-from .models import Run
+from .models import Run, ThreadAndRunCreateParams
 from .service import RunService
 
-router = APIRouter(prefix="/threads/{thread_id}/runs", tags=["runs"])
+router = APIRouter(tags=["runs"])
 
 
-@router.post("")
+@router.post("/threads/{thread_id}/runs")
 async def create_run(
     thread_id: Annotated[ThreadId, Path(...)],
     params: RunCreateParams,
@@ -53,7 +53,40 @@ async def _run_async_generator() -> None:
     return JSONResponse(status_code=status.HTTP_201_CREATED, content=run.model_dump())
 
 
-@router.get("/{run_id}")
+@router.post("/runs")
+async def create_thread_and_run(
+    params: ThreadAndRunCreateParams,
+    background_tasks: BackgroundTasks,
+    thread_facade: ThreadFacade = ThreadFacadeDep,
+) -> Response:
+    stream = params.stream
+    run, async_generator = await thread_facade.create_thread_and_run(params)
+    if stream:
+
+        async def sse_event_stream() -> AsyncGenerator[str, None]:
+            async for event in async_generator:
+                data = (
+                    event.data.model_dump_json()
+                    if isinstance(event.data, BaseModel)
+                    else event.data
+                )
+                yield f"event: {event.event}\ndata: {data}\n\n"
+
+        return StreamingResponse(
+            status_code=status.HTTP_201_CREATED,
+            content=sse_event_stream(),
+            media_type="text/event-stream",
+        )
+
+    async def _run_async_generator() -> None:
+        async for _ in async_generator:
+            pass
+
+    background_tasks.add_task(_run_async_generator)
+    return JSONResponse(status_code=status.HTTP_201_CREATED, content=run.model_dump())
+
+
+@router.get("/threads/{thread_id}/runs/{run_id}")
 def retrieve_run(
     thread_id: Annotated[ThreadId, Path(...)],
     run_id: Annotated[RunId, Path(...)],
@@ -62,7 +95,7 @@ def retrieve_run(
     return run_service.retrieve(thread_id, run_id)
 
 
-@router.get("")
+@router.get("/threads/{thread_id}/runs")
 def list_runs(
     thread_id: Annotated[ThreadId, Path(...)],
     query: ListQuery = ListQueryDep,
@@ -71,7 +104,7 @@ def list_runs(
     return thread_facade.list_runs(thread_id, query=query)
 
 
-@router.post("/{run_id}/cancel")
+@router.post("/threads/{thread_id}/runs/{run_id}/cancel")
 def cancel_run(
     thread_id: Annotated[ThreadId, Path(...)],
     run_id: Annotated[RunId, Path(...)],

src/askui/chat/api/threads/facade.py

@@ -3,7 +3,7 @@
 from askui.chat.api.messages.models import Message, MessageCreateParams
 from askui.chat.api.messages.service import MessageService
 from askui.chat.api.models import ThreadId
-from askui.chat.api.runs.models import Run, RunCreateParams
+from askui.chat.api.runs.models import Run, RunCreateParams, ThreadAndRunCreateParams
 from askui.chat.api.runs.runner.events.events import Events
 from askui.chat.api.runs.service import RunService
 from askui.chat.api.threads.service import ThreadService
@@ -43,6 +43,13 @@ async def create_run(
         self._ensure_thread_exists(thread_id)
         return await self._run_service.create(thread_id, params)
 
+    async def create_thread_and_run(
+        self, params: ThreadAndRunCreateParams
+    ) -> tuple[Run, AsyncGenerator[Events, None]]:
+        """Create a thread and a run, ensuring the thread exists first."""
+        thread = self._thread_service.create(params.thread)
+        return await self._run_service.create(thread.id, params)
+
     def list_messages(
         self, thread_id: ThreadId, query: ListQuery
     ) -> ListResponse[Message]: