Unable to run PydanticAI

Hi, I have been trying for days to deploy the python code with pydanticai. Locally, everything works well, but on vercel, I am constantly getting FUNCTION_INVOCATION_FAILED error. when I check the log, I see the following:

Traceback (most recent call last):
File "/var/task/vc__handler__python.py", line 14, in <module>
__vc_spec.loader.exec_module(__vc_module)
File "<frozen importlib._bootstrap_external>", line 995, in exec_module
File "<frozen importlib._bootstrap>", line 488, in _call_with_frames_removed
File "/var/task/src/main_vercel.py", line 1, in <module>
from fastapi import FastAPI, Request, WebSocket
File "/var/task/fastapi/__init__.py", line 7, in <module>
from .applications import FastAPI as FastAPI
File "/var/task/fastapi/applications.py", line 16, in <module>
from fastapi import routing
File "/var/task/fastapi/routing.py", line 24, in <module>
from fastapi import params
File "/var/task/fastapi/params.py", line 5, in <module>
from fastapi.openapi.models import Example
File "/var/task/fastapi/openapi/models.py", line 4, in <module>
from fastapi._compat import (
File "/var/task/fastapi/_compat.py", line 21, in <module>
from fastapi.exceptions import RequestErrorModel
File "/var/task/fastapi/exceptions.py", line 3, in <module>
from pydantic import BaseModel, create_model
File "<frozen importlib._bootstrap>", line 1412, in _handle_fromlist
File "/var/task/pydantic/__init__.py", line 421, in __getattr__
module = import_module(module_name, package=package)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/var/lang/lib/python3.12/importlib/__init__.py", line 90, in import_module
return _bootstrap._gcd_import(name[level:], package, level)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/var/task/pydantic/main.py", line 34, in <module>
from ._internal import (
File "/var/task/pydantic/_internal/_model_construction.py", line 25, in <module>
from ._generate_schema import GenerateSchema
File "/var/task/pydantic/_internal/_generate_schema.py", line 42, in <module>
from uuid import UUID
File "/var/task/uuid.py", line 138
if not 0 <= time_low < 1<<32L:
^
SyntaxError: invalid decimal literal
Python process exited with exit status: 1. The logs above can help with debugging the issue.

the main_vercel.py script is the following:

from fastapi import FastAPI, Request, WebSocket
from fastapi.middleware.cors import CORSMiddleware
from dotenv import load_dotenv
import os
from src.routers import (
    image_router, 
    auth_router, 
    financial_stats_router, 
    receipt, 
    financial_analytics_router, 
    store, 
    tags,
    spending_sheet,
    categories,
    user_settings_router
)
from src.websockets.websocket_manager import manager
from src.database.db_utils import init_db
from src.type_defs.dynamic_types import initialize_types
from contextlib import asynccontextmanager

# Load environment variables
load_dotenv()

# Define lifespan first
@asynccontextmanager
async def lifespan(app: FastAPI):
    """Lifespan context manager for FastAPI app"""
    # Startup
    await init_db()
    await initialize_types()
    yield
    # Shutdown
    # Add any cleanup code here

# Create FastAPI app instance ONCE with lifespan
app = FastAPI(
    title="Expenzor API",
    description="Backend API for Expenzor expense tracking application",
    version="1.0.0",
    lifespan=lifespan
)

# Configure CORS
origins = os.getenv("ALLOWED_ORIGINS", https://expenzor.com,https://www.expenzor.com").split(",")

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],  # Allows all methods
    allow_headers=["*"],  # Allows all headers
    expose_headers=["*"]  # Exposes all headers
)

# Add request logging middleware for production
@app.middleware("http")
async def log_requests(request: Request, call_next):
    # Log basic request info
    print(f"\n--- Request ---")
    print(f"Method: {request.method}")
    print(f"URL: {request.url}")
    
    # Skip body logging for multipart/form-data (file uploads)
    if not request.headers.get("content-type", "").startswith("multipart/form-data"):
        body = await request.body()
        try:
            print(f"Body: {body.decode()}")
        except UnicodeDecodeError:
            print("Body: <binary data>")
    else:
        print("Body: <file upload>")
    
    response = await call_next(request)
    print(f"Response status code: {response.status_code}")
    return response

# WebSocket endpoint
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
    print(f"WebSocket connection attempt from client: {client_id}")
    await manager.connect(websocket, client_id)
    try:
        while True:
            data = await websocket.receive_text()
            print(f"Received message from client {client_id}: {data}")
    except Exception as e:
        print(f"WebSocket error for client {client_id}: {e}")
    finally:
        print(f"WebSocket connection closed for client {client_id}")
        await manager.disconnect(websocket, client_id)

# Include routers
app.include_router(image_router)
app.include_router(auth_router)
app.include_router(financial_stats_router)
app.include_router(receipt.router)
app.include_router(financial_analytics_router)
app.include_router(store.router)
app.include_router(tags.router)
app.include_router(spending_sheet.router)
app.include_router(categories.router)
app.include_router(user_settings_router)

# Root endpoint
@app.get("/")
async def root():
    return {"message": "Welcome to Expenzor API"}

# Health check endpoint
@app.get("/health")
async def health_check():
    return {"status": "ok"}

if __name__ == "__main__":
    import uvicorn
    import sys
    
    # Determine environment
    env = os.getenv("ENV", "development")
    
    # Configure server settings based on environment
    host = "0.0.0.0"
    port = int(os.getenv("PORT", "8000"))
    reload_enabled = env == "development"
    
    # Check if running in debug mode
    is_debug = sys.gettrace() is not None
    
    if is_debug:
        print("Debug mode detected. Use the 'FastAPI' launch configuration to debug with the server running.")
    else:
        print(f"Starting server in {env} mode")
        print(f"Host: {host}")
        print(f"Port: {port}")
        print(f"Auto-reload: {'enabled' if reload_enabled else 'disabled'}")
        
        if reload_enabled:
            # When reload is enabled, we need to use the import string
            uvicorn.run(
                "src.main:app",
                host=host,
                port=port,
                reload=True,
                log_level="info"
            )
        else:
            # When reload is disabled, we can use the app instance directly
            uvicorn.run(
                app,
                host=host,
                port=port,
                reload=False,
                log_level="info"
            )

these are essential imports in some other script:

from dataclasses import dataclass
import asyncio
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext
from uuid import UUID, uuid4

this is the requirements.txt file:

aiohappyeyeballs==2.4.4
aiohttp==3.11.11
aiosignal==1.3.2
annotated-types==0.7.0
anyio==4.8.0
asyncpg==0.29.0
attrs==25.1.0
bcrypt==4.2.1
boto3==1.29.0
botocore==1.32.7
cachetools==5.5.1
certifi==2025.1.31
cffi==1.17.1
charset-normalizer==3.4.1
click==8.1.8
colorama==0.4.6
cryptography==44.0.0
deprecation==2.1.0
distro==1.9.0
dnspython==2.7.0
ecdsa==0.19.0
email_validator==2.2.0
eval_type_backport==0.2.2
fastapi==0.115.0
fastavro==1.10.0
filelock==3.17.0
frozenlist==1.5.0
fsspec==2025.2.0
google-auth==2.38.0
greenlet==3.1.1
griffe==1.5.6
h11==0.14.0
h2==4.2.0
hpack==4.1.0
httpcore==1.0.7
httptools==0.6.4
httpx==0.28.1
httpx-sse==0.4.0
huggingface-hub==0.28.1
hyperframe==6.1.0
idna==3.10
jiter==0.8.2
jmespath==1.0.1
jsonpath-python==1.0.6
logfire-api==3.5.0
multidict==6.1.0
mypy-extensions==1.0.0
ollama==0.4.7
openai==1.61.0
packaging==24.2
passlib==1.7.4
Pillow==10.1.0
postgrest==0.19.3
propcache==0.2.1
pyasn1==0.6.1
pyasn1_modules==0.4.1
pycparser==2.22
pydantic-ai==0.0.21
pydantic_core==2.27.2
python-dateutil==2.9.0.post0
python-dotenv==1.0.0
python-jose==3.3.0
python-magic==0.4.27
python-multipart==0.0.6
PyYAML==6.0.2
realtime==2.3.0
requests==2.32.3
rsa==4.9
s3transfer==0.7.0
six==1.17.0
sniffio==1.3.1
SQLAlchemy==2.0.23
storage3==0.11.3
StrEnum==0.4.15
supabase==2.12.0
supafunc==0.9.3
tokenizers==0.21.0
tqdm==4.67.1
types-requests==2.32.0.20241016
typing-inspect==0.9.0
typing_extensions==4.12.2
urllib3==2.0.7
uvicorn==0.30.6
uvloop==0.21.0
watchfiles==1.0.4
websockets==12.0
yarl==1.18.3

I would be thankful if anyone helped me with solving the issue. I suspect there might be an issue with pydantic v2.x but I can’t use older version due to other dependencies.

Hi, Niko!

Could you try the troubleshooting steps we’ve outlined here?

Thanks for help. I implemented additional try/catch statements and actually the issue was with UUID. Strangely, locally, everything worked well, but on Vercel, I was constantly getting error.

In the end, I wrote my uuid patc:

"""
Patched UUID module for Python 3.12 compatibility.
This module provides a drop-in replacement for the standard uuid module.
"""
import os
import random
from typing import Optional, Union, Tuple

# Constants from the original UUID module
NAMESPACE_DNS = '6ba7b810-9dad-11d1-80b4-00c04fd430c8'
NAMESPACE_URL = '6ba7b811-9dad-11d1-80b4-00c04fd430c8'
NAMESPACE_OID = '6ba7b812-9dad-11d1-80b4-00c04fd430c8'
NAMESPACE_X500 = '6ba7b814-9dad-11d1-80b4-00c04fd430c8'

class UUID:
    def __init__(self, hex: Optional[str] = None, bytes_: Optional[bytes] = None, 
                 int_: Optional[int] = None, version: Optional[int] = None):
        if [hex, bytes_, int_].count(None) != 2:
            raise TypeError('Need exactly one of hex, bytes_, or int_')
        
        if hex is not None:
            hex = hex.replace('urn:', '').replace('uuid:', '')
            hex = hex.strip('{}').replace('-', '')
            if len(hex) != 32:
                raise ValueError('Invalid UUID string')
            self._int = int(hex, 16)
        elif bytes_ is not None:
            if len(bytes_) != 16:
                raise ValueError('Invalid UUID bytes')
            self._int = int.from_bytes(bytes_, byteorder='big')
        elif int_ is not None:
            self._int = int_
        
        if version is not None:
            if not 1 <= version <= 5:
                raise ValueError('Invalid version number')
            # Set version
            self._int &= ~(0xf000)
            self._int |= version << 12
    
    @property
    def bytes(self) -> bytes:
        return self._int.to_bytes(16, byteorder='big')
    
    @property
    def hex(self) -> str:
        return '%032x' % self._int
    
    @property
    def int(self) -> int:
        return self._int
    
    def __str__(self) -> str:
        hex = self.hex
        return f'{hex[:8]}-{hex[8:12]}-{hex[12:16]}-{hex[16:20]}-{hex[20:]}'
    
    def __repr__(self) -> str:
        return f'UUID({str(self)})'
    
    def __eq__(self, other: object) -> bool:
        if isinstance(other, UUID):
            return self._int == other._int
        return NotImplemented
    
    def __hash__(self) -> int:
        return hash(self._int)

def uuid4() -> UUID:
    """Generate a random UUID (version 4)."""
    # Get 16 random bytes
    random_bytes = os.urandom(16)
    # Set version 4 and variant bits
    random_bytes = bytearray(random_bytes)
    random_bytes[6] = (random_bytes[6] & 0x0f) | 0x40  # version 4
    random_bytes[8] = (random_bytes[8] & 0x3f) | 0x80  # variant
    return UUID(bytes_=bytes(random_bytes))

# Expose the main functions and constants
__all__ = ['UUID', 'uuid4', 'NAMESPACE_DNS', 'NAMESPACE_URL', 'NAMESPACE_OID', 'NAMESPACE_X500'] 

and in the main_vercel.py file, I have patched the original uuid with patched:

from src.utils.uuid_patch import UUID, uuid4
    import sys
    sys.modules['uuid'] = sys.modules['src.utils.uuid_patch']
2 Likes

This topic was automatically closed 7 days after the last reply. New replies are no longer allowed.