type lint

Author: chriscarrollsmith

.gitignore

@@ -5,4 +5,6 @@
 __pycache__
 *.pyc
 .specstory
-.mypy_cache
+.mypy_cache
+.cursorrules
+.repomix-output.txt

main.py

@@ -1,11 +1,12 @@
 import os
 import logging
+from typing import List, Dict, Any, Optional
 from dotenv import load_dotenv
 from contextlib import asynccontextmanager
 from fastapi import FastAPI, Request
 from fastapi.staticfiles import StaticFiles
 from fastapi.templating import Jinja2Templates
-from fastapi.responses import RedirectResponse
+from fastapi.responses import RedirectResponse, Response
 from routers import chat, files, setup
 from utils.threads import create_thread
 from fastapi.exceptions import HTTPException
@@ -33,7 +34,7 @@ async def lifespan(app: FastAPI):
 templates = Jinja2Templates(directory="templates")
 
 @app.exception_handler(Exception)
-async def general_exception_handler(request: Request, exc: Exception):
+async def general_exception_handler(request: Request, exc: Exception) -> Response:
     logger.error(f"Unhandled error: {exc}")
     return templates.TemplateResponse(
         "error.html",
@@ -42,7 +43,7 @@ async def general_exception_handler(request: Request, exc: Exception):
     )
 
 @app.exception_handler(HTTPException)
-async def http_exception_handler(request: Request, exc: HTTPException):
+async def http_exception_handler(request: Request, exc: HTTPException) -> Response:
     logger.error(f"HTTP error: {exc.detail}")
     return templates.TemplateResponse(
         "error.html",
@@ -54,23 +55,30 @@ async def http_exception_handler(request: Request, exc: HTTPException):
 # TODO: Implement some kind of thread id storage or management logic to allow
 # user to load an old thread, delete an old thread, etc. instead of start new
 @app.get("/")
-async def read_home(request: Request, thread_id: str = None, messages: list = []):
+async def read_home(
+    request: Request,
+    thread_id: Optional[str] = None,
+    messages: List[Dict[str, Any]] = []
+) -> Response:
     logger.info("Home page requested")
 
     # Check if environment variables are missing
     load_dotenv(override=True)
-    if not os.getenv("OPENAI_API_KEY") or not os.getenv("ASSISTANT_ID"):
+    openai_api_key = os.getenv("OPENAI_API_KEY")
+    assistant_id = os.getenv("ASSISTANT_ID")
+    
+    if not openai_api_key or not assistant_id:
         return RedirectResponse(url=app.url_path_for("read_setup"))
 
     # Create a new assistant chat thread if no thread ID is provided
     if not thread_id or thread_id == "None" or thread_id == "null":
-        thread_id: str = await create_thread()
+        thread_id = await create_thread()
 
     return templates.TemplateResponse(
         "index.html",
         {
             "request": request,
-            "assistant_id": os.getenv("ASSISTANT_ID"),
+            "assistant_id": assistant_id,
             "messages": messages,
             "thread_id": thread_id
         }

routers/chat.py

@@ -1,7 +1,7 @@
 import logging
 import time
 from datetime import datetime
-from typing import Any, AsyncGenerator
+from typing import Any, AsyncGenerator, Dict, List, Optional, Union, cast
 from fastapi.templating import Jinja2Templates
 from fastapi import APIRouter, Form, Depends, Request
 from fastapi.responses import StreamingResponse, HTMLResponse
@@ -38,10 +38,10 @@
 
 # Utility function for submitting tool outputs to the assistant
 class ToolCallOutputs(BaseModel):
-    tool_outputs: Any
+    tool_outputs: Dict[str, Any]
     runId: str
 
-async def post_tool_outputs(client: AsyncOpenAI, data: dict, thread_id: str):
+async def post_tool_outputs(client: AsyncOpenAI, data: Dict[str, Any], thread_id: str) -> AsyncAssistantStreamManager:
     """
     data is expected to be something like
     {
@@ -55,7 +55,7 @@ async def post_tool_outputs(client: AsyncOpenAI, data: dict, thread_id: str):
     try:
         outputs_list = [
             ToolOutput(
-                output=data["tool_outputs"]["output"],
+                output=str(data["tool_outputs"]["output"]),
                 tool_call_id=data["tool_outputs"]["tool_call_id"]
             )
         ]
@@ -124,14 +124,14 @@ async def handle_assistant_stream(
         templates: Jinja2Templates,
         logger: logging.Logger,
         stream_manager: AsyncAssistantStreamManager,
-        step_id: int = 0
-    ) -> AsyncGenerator:
+        step_id: str = ""
+    ) -> AsyncGenerator[Union[Dict[str, Any], str], None]:
         """
         Async generator to yield SSE events.
         We yield a final 'metadata' dictionary event once we're done.
         """
-        required_action: RequiredAction | None = None
-        run_requires_action_event: ThreadRunRequiresAction | None = None
+        required_action: Optional[RequiredAction] = None
+        run_requires_action_event: Optional[ThreadRunRequiresAction] = None
 
         event_handler: AsyncAssistantEventHandler
         async with stream_manager as event_handler:
@@ -149,11 +149,13 @@ async def handle_assistant_stream(
                     )
                     time.sleep(0.25)  # Give the client time to render the message
 
-                if isinstance(event, ThreadMessageDelta):
-                    yield sse_format(
-                        f"textDelta{step_id}",
-                        event.data.delta.content[0].text.value
-                    )
+                if isinstance(event, ThreadMessageDelta) and event.data.delta.content:
+                    content = event.data.delta.content[0]
+                    if hasattr(content, 'text') and content.text and content.text.value:
+                        yield sse_format(
+                            f"textDelta{step_id}",
+                            content.text.value
+                        )
 
                 if isinstance(event, ThreadRunStepCreated) and event.data.type == "tool_calls":
                     step_id = event.data.id
@@ -167,47 +169,50 @@ async def handle_assistant_stream(
                     )
                     time.sleep(0.25)  # Give the client time to render the message
 
-                if isinstance(event, ThreadRunStepDelta) and event.data.delta.step_details.type == "tool_calls":
-                    tool_call = event.data.delta.step_details.tool_calls[0]
-                    
-                    # Handle function tool calls
-                    if tool_call.type == "function":
-                        if tool_call.function.name:
-                            yield sse_format(
-                                f"toolDelta{step_id}",
-                                tool_call.function.name + "<br>"
-                            )
-                        elif tool_call.function.arguments:
-                            yield sse_format(
-                                f"toolDelta{step_id}",
-                                tool_call.function.arguments
-                            )
-                    
-                    # Handle code interpreter tool calls
-                    elif tool_call.type == "code_interpreter":
-                        if tool_call.code_interpreter.input:
-                            yield sse_format(
-                                f"toolDelta{step_id}",
-                                f"{tool_call.code_interpreter.input}"
-                            )
-                        if tool_call.code_interpreter.outputs:
-                            for output in tool_call.code_interpreter.outputs:
-                                if output.type == "logs":
-                                    yield sse_format(
-                                        f"toolDelta{step_id}",
-                                        f"{output.logs}"
-                                    )
-                                elif output.type == "image":
-                                    yield sse_format(
-                                        f"toolDelta{step_id}",
-                                        f"{output.image.file_id}"
-                                    )
+                if isinstance(event, ThreadRunStepDelta) and event.data.delta.step_details and event.data.delta.step_details.type == "tool_calls":
+                    tool_calls = event.data.delta.step_details.tool_calls
+                    if tool_calls:
+                        # TODO: Support parallel function calling
+                        tool_call = tool_calls[0]
+
+                        # Handle function tool call
+                        if tool_call.type == "function":
+                            if tool_call.function and tool_call.function.name:
+                                yield sse_format(
+                                    f"toolDelta{step_id}",
+                                    tool_call.function.name + "<br>"
+                                )
+                            if tool_call.function and tool_call.function.arguments:
+                                yield sse_format(
+                                    f"toolDelta{step_id}",
+                                    tool_call.function.arguments
+                                )
+                        
+                        # Handle code interpreter tool calls
+                        elif tool_call.type == "code_interpreter":
+                            if tool_call.code_interpreter and tool_call.code_interpreter.input:
+                                yield sse_format(
+                                    f"toolDelta{step_id}",
+                                    str(tool_call.code_interpreter.input)
+                                )
+                            if tool_call.code_interpreter and tool_call.code_interpreter.outputs:
+                                for output in tool_call.code_interpreter.outputs:
+                                    if output.type == "logs" and output.logs:
+                                        yield sse_format(
+                                            f"toolDelta{step_id}",
+                                            str(output.logs)
+                                        )
+                                    elif output.type == "image" and output.image and output.image.file_id:
+                                        yield sse_format(
+                                            f"toolDelta{step_id}",
+                                            str(output.image.file_id)
+                                        )
 
                 # If the assistant run requires an action (a tool call), break and handle it
                 if isinstance(event, ThreadRunRequiresAction):
                     required_action = event.data.required_action
                     run_requires_action_event = event
-                    if required_action.submit_tool_outputs:
+                    if required_action and required_action.submit_tool_outputs:
                         break
 
                 if isinstance(event, ThreadRunCompleted):
@@ -221,45 +226,50 @@ async def handle_assistant_stream(
             "run_requires_action_event": run_requires_action_event
         }
 
-    async def event_generator():
+    async def event_generator() -> AsyncGenerator[str, None]:
         """
         Main generator for SSE events. We call our helper function to handle the assistant
         stream, and if the assistant requests a tool call, we do it and then re-run the stream.
         """
-        step_id = 0
-        initial_manager = client.beta.threads.runs.stream(
+        step_id: str = ""
+        stream_manager: AsyncAssistantStreamManager[AsyncAssistantEventHandler] = client.beta.threads.runs.stream(
             assistant_id=assistant_id,
             thread_id=thread_id,
             parallel_tool_calls=False
         )
 
-        stream_manager = initial_manager
         while True:
+            event: dict[str, Any] | str
             async for event in handle_assistant_stream(templates, logger, stream_manager, step_id):
                 # Detect the special "metadata" event at the end of the generator
                 if isinstance(event, dict) and event.get("type") == "metadata":
-                    required_action: RequiredAction | None = event["required_action"]
-                    step_id: int = event["step_id"]
-                    run_requires_action_event: ThreadRunRequiresAction | None = event["run_requires_action_event"]
+                    required_action = cast(Optional[RequiredAction], event.get("required_action"))
+                    step_id = cast(str, event.get("step_id", ""))
+                    run_requires_action_event = cast(Optional[ThreadRunRequiresAction], event.get("run_requires_action_event"))
 
                     # If the assistant still needs a tool call, do it and then re-stream
-                    if required_action and required_action.submit_tool_outputs:
+                    if required_action and required_action.submit_tool_outputs and required_action.submit_tool_outputs.tool_calls:
                         for tool_call in required_action.submit_tool_outputs.tool_calls:
                             if tool_call.type == "function":
                                 try:
                                     args = json.loads(tool_call.function.arguments)
                                     location = args.get("location", "Unknown")
-                                    dates = args.get("dates", [datetime.today()])
+                                    dates_raw = args.get("dates", [datetime.today().strftime("%Y-%m-%d")])
+                                    dates = [
+                                        datetime.strptime(d, "%Y-%m-%d") if isinstance(d, str) else d 
+                                        for d in dates_raw
+                                    ]
                                 except Exception as err:
                                     logger.error(f"Failed to parse function arguments: {err}")
                                     location = "Unknown"
+                                    dates = [datetime.today()]
 
                                 try:
-                                    weather_output: list[dict] = get_weather(location, dates)
+                                    weather_output: list = get_weather(location, dates)
                                     logger.info(f"Weather output: {weather_output}")
 
                                     # Render the weather widget
-                                    weather_widget_html: str = templates.get_template(
+                                    weather_widget_html = templates.get_template(
                                         "components/weather-widget.html"
                                     ).render(
                                         reports=weather_output
@@ -273,7 +283,7 @@ async def event_generator():
                                             "output": str(weather_output),
                                             "tool_call_id": tool_call.id
                                         },
-                                        "runId": run_requires_action_event.data.id,
+                                        "runId": run_requires_action_event.data.id if run_requires_action_event else "",
                                     }
                                 except Exception as err:
                                     error_message = f"Failed to get weather output: {err}"
@@ -284,24 +294,24 @@ async def event_generator():
                                             "output": error_message,
                                             "tool_call_id": tool_call.id
                                         },
-                                        "runId": run_requires_action_event.data.id,
+                                        "runId": run_requires_action_event.data.id if run_requires_action_event else "",
                                     }
 
-                        # Afterwards, create a fresh stream_manager for the next iteration
-                        new_stream_manager: AsyncAssistantStreamManager = await post_tool_outputs(
-                            client,
-                            data_for_tool,
-                            thread_id
-                        )
-                        stream_manager = new_stream_manager
-                        # proceed to rerun the loop
-                        break
+                                # Afterwards, create a fresh stream_manager for the next iteration
+                                new_stream_manager = await post_tool_outputs(
+                                    client,
+                                    data_for_tool,
+                                    thread_id
+                                )
+                                stream_manager = new_stream_manager
+                                # proceed to rerun the loop
+                                break
                     else:
                         # No more tool calls needed; we're done streaming
                         return
                 else:
                     # Normal SSE events: yield them to the client
-                    yield event
+                    yield str(event)
 
     return StreamingResponse(
         event_generator(),