derp.ai – AI Client

AI client wrapping OpenAI-compatible providers.

class derp.ai.AIClient[source]

Bases: object

Async AI client wrapping several providers.

Example:

config = AIConfig(api_key="...")
ai = AIClient(config)
response = await ai.chat(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello"}],
)
__init__(config)[source]
Parameters:

config (AIConfig)

async connect()[source]
Return type:

None

async disconnect()[source]
Return type:

None

async chat(model, *, messages, tools=(), **kwargs)[source]

Create a chat completion.

Parameters:

  • model (str) – Model ID to use.

  • messages (list[dict[str, Any]]) – List of message dicts.

  • tools (Sequence[type[Tool]]) – Optional list of Tool subclasses.

  • **kwargs (Any) – Additional arguments forwarded to the API.

Returns:

ChatResponse with content, usage, and protocol adapters.

Return type:

ChatResponse

async stream_chat(model, *, messages, tools=(), **kwargs)[source]

Create a streaming chat completion.

Parameters:

  • model (str) – Model ID to use.

  • messages (list[dict[str, Any]]) – List of message dicts.

  • tools (Sequence[type[Tool]]) – Optional list of Tool subclasses.

  • **kwargs (Any) – Additional arguments forwarded to the API.

Yields:

ChatChunk for each text delta. The final chunk includes parsed tool_calls when the model invokes tools.

Return type:

AsyncIterator[ChatChunk]

async stream_agent(model, *, messages, tools=(), tool_args=(), max_turns=10, **kwargs)[source]

Stream a chat completion loop, auto-executing tool calls.

Streams via stream_chat() in a loop. Text deltas are yielded as they arrive. When the model returns tool calls, each tool is executed via its run() method, results are appended as tool messages, and the next round starts automatically.

The loop continues until the model returns a text response (no tool calls) or max_turns is reached.

Parameters:

  • model (str) – Model ID to use.

  • messages (list[dict[str, Any]]) – List of message dicts (mutated in place).

  • tools (Sequence[type[Tool]]) – Tool subclasses available to the model.

  • tool_args (Sequence[Any]) – Extra positional args forwarded to each Tool.run() call (e.g. request-scoped state).

  • max_turns (int) – Maximum number of tool-call round-trips.

  • **kwargs (Any) – Additional arguments forwarded to the API.

Yields:

ChatChunk for each text delta across all turns.

Return type:

AsyncIterator[ChatChunk]

async fal_submit(app, *, inputs, start_timeout=10.0)[source]

Submit a job to a Fal application.

Parameters:

  • app (str) – Fal application name.

  • inputs (dict[str, Any]) – Inputs to the model.

  • start_timeout (float) – Start timeout in seconds. Default is 10 seconds.

Returns:

Request ID of the submitted task.

Return type:

str

async fal_call(app, *, inputs, poll_interval=2.0, timeout=60.0, start_timeout=10.0)[source]

Submit a fal job and wait for the result.

Convenience method combining fal_submit(), fal_poll(), and fal_get() into a single call.

Parameters:

  • app (str) – Fal application name.

  • inputs (dict[str, Any]) – Inputs to the model.

  • poll_interval (float) – Seconds between status polls. Default is 2.

  • timeout (float) – Maximum seconds to wait. Default is 60.

  • start_timeout (float) – Start timeout in seconds. Default is 10.

Returns:

Result dict from the completed job.

Raises:

  • FalJobFailedError – If the job fails.

  • TimeoutError – If the job does not complete within timeout.

Return type:

dict[str, Any]

async fal_poll(app, request_id)[source]

Poll the status of a fal job.

Parameters:

  • app (str) – Fal application name.

  • request_id (str) – Request ID returned by fal_submit.

Returns:

JobStatus with the current state of the job.

Return type:

JobStatus

async fal_get(app, request_id)[source]

Get the result of a fal job.

Parameters:

  • app (str) – Fal application name.

  • request_id (str) – Request ID returned by fal_submit.

Returns:

Result of the job as a dict.

Return type:

dict[str, Any]

async fal_cancel(app, request_id)[source]

Cancel a fal job.

Parameters:

  • app (str) – Fal application name.

  • request_id (str) – Request ID returned by fal_submit.

Returns:

CancelResult with the cancellation state and job state.

Raises:

  • FalJobAlreadyCompletedError – If the job already completed.

  • FalJobNotFoundError – If the job was not found.

Return type:

CancelResult

async modal_call(endpoint, *, inputs, timeout=30.0)[source]

Call a Modal endpoint.

Parameters:

  • endpoint (str) – Modal endpoint name.

  • inputs (dict[str, Any]) – Inputs to the endpoint.

  • timeout (float) – Timeout in seconds. Default is 30 seconds.

Returns:

Result of the endpoint as a dict.

Return type:

dict[str, Any]

class derp.ai.AIConfig[source]

Bases: _StrictModel

AI configuration for OpenAI-compatible providers.

api_key: str
base_url: str | None
fal_api_key: str | None
modal: ModalConfig | None
model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class derp.ai.ChatChunk[source]

Bases: BaseModel

A single chunk from a streaming chat completion.

delta: str
role: str
model: str | None
finish_reason: str | None
usage: Usage | None
tool_calls: list[ToolCall]
is_first: bool
is_last: bool
tool_event: ToolEventType | None
tool_call_id: str | None
tool_name: str | None
tool_input: dict[str, Any] | None
tool_output: Any
vercel_ai_json(*, message_id, stream_id='text-1')[source]

Format as Vercel AI SDK events.

Includes lifecycle events when is_first/is_last are set. When tool_event is set, emits tool lifecycle events instead.

Parameters:

  • message_id (str)

  • stream_id (str)

Return type:

list[SSEEvent]

tanstack_ai_json(*, message_id, run_id=None)[source]

Format as TanStack AG-UI events.

Includes lifecycle events when is_first/is_last are set. When tool_event is set, emits AG-UI tool call events instead.

Parameters:

  • message_id (str)

  • run_id (str | None)

Return type:

list[SSEEvent]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class derp.ai.ChatResponse[source]

Bases: BaseModel

Non-streaming chat completion response.

content: str
role: str
model: str
usage: Usage | None
finish_reason: str
tool_calls: list[ToolCall]
vercel_ai_json(*, message_id=None)[source]

Format as Vercel AI SDK data stream protocol events.

Parameters:

message_id (str | None)

Return type:

list[SSEEvent]

tanstack_ai_json(*, run_id=None, message_id=None)[source]

Format as TanStack AG-UI protocol events.

Parameters:

  • run_id (str | None)

  • message_id (str | None)

Return type:

list[SSEEvent]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class derp.ai.JobState[source]

Bases: StrEnum

State of an async AI job.

QUEUED = 'queued'
IN_PROGRESS = 'in_progress'
COMPLETED = 'completed'
FAILED = 'failed'
UNKNOWN = 'unknown'
__new__(value)
class derp.ai.JobStatus[source]

Bases: BaseModel

Status of an async AI job (e.g. fal image generation).

Wraps fal’s Queued/InProgress/Completed statuses into a single model.

state: JobState
position: int | None
logs: list[dict[str, Any]] | None
metrics: dict[str, Any] | None
error: str | None
error_type: str | None
property is_queued: bool
property is_in_progress: bool
property is_completed: bool
property is_failed: bool
property is_done: bool
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class derp.ai.Tool[source]

Bases: BaseModel, ABC

Base class for defining AI tools.

Subclass with typed fields for parameters, a docstring for the description, and implement run():

class GetWeather(Tool):
    """Get the current weather for a city."""
    city: str
    unit: str = "celsius"

    async def run(self) -> str:
        return f"22° {self.unit}"

Pass the class (not an instance) to chat() or run():

response = await ai.chat(model="gpt-4o", messages=msgs, tools=[GetWeather])
abstractmethod async run(*args, **kwargs)[source]

Execute the tool. Override in subclasses.

Parameters:
Return type:

Any

classmethod function_name()[source]

The function name sent to the LLM (snake_cased class name).

Return type:

str

classmethod openai_schema()[source]

Generate the OpenAI function-tool JSON schema.

Return type:

dict[str, Any]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class derp.ai.ToolCall[source]

Bases: BaseModel

A parsed tool call returned by the LLM.

id: str
function_name: str
arguments: str
args: Any
async run(*args, **kwargs)[source]

Execute the tool call. Only works when args is a Tool instance.

Extra positional and keyword arguments are forwarded to Tool.run(), allowing request-scoped state to be injected without closures.

Parameters:
Return type:

Any

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class derp.ai.ToolEventType[source]

Bases: StrEnum

Type of tool lifecycle event emitted during agentic streaming.

INPUT_START = 'input_start'
INPUT_AVAILABLE = 'input_available'
OUTPUT_AVAILABLE = 'output_available'
__new__(value)
class derp.ai.Usage[source]

Bases: BaseModel

Token usage statistics.

prompt_tokens: int
completion_tokens: int
total_tokens: int
vercel_ai_json()[source]

Format as Vercel AI SDK usage object.

Return type:

dict[str, int]

tanstack_ai_json()[source]

Format as TanStack AG-UI usage object.

Return type:

dict[str, int]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Provider-agnostic AI response models with protocol adapters.

class derp.ai.models.SSEEvent[source]

Bases: dict[str, Any]

A single SSE event. Behaves like a dict with a dump() method.

dump()[source]

Serialize as an SSE data line: data: {...}\n\n.

Return type:

str

class derp.ai.models.SSEDone[source]

Bases: SSEEvent

Terminal SSE event signaling end of stream.

dump()[source]

Serialize as an SSE data line: data: {...}\n\n.

Return type:

str

class derp.ai.models.Usage[source]

Bases: BaseModel

Token usage statistics.

prompt_tokens: int
completion_tokens: int
total_tokens: int
vercel_ai_json()[source]

Format as Vercel AI SDK usage object.

Return type:

dict[str, int]

tanstack_ai_json()[source]

Format as TanStack AG-UI usage object.

Return type:

dict[str, int]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class derp.ai.models.Tool[source]

Bases: BaseModel, ABC

Base class for defining AI tools.

Subclass with typed fields for parameters, a docstring for the description, and implement run():

class GetWeather(Tool):
    """Get the current weather for a city."""
    city: str
    unit: str = "celsius"

    async def run(self) -> str:
        return f"22° {self.unit}"

Pass the class (not an instance) to chat() or run():

response = await ai.chat(model="gpt-4o", messages=msgs, tools=[GetWeather])
abstractmethod async run(*args, **kwargs)[source]

Execute the tool. Override in subclasses.

Parameters:
Return type:

Any

classmethod function_name()[source]

The function name sent to the LLM (snake_cased class name).

Return type:

str

classmethod openai_schema()[source]

Generate the OpenAI function-tool JSON schema.

Return type:

dict[str, Any]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class derp.ai.models.ToolCall[source]

Bases: BaseModel

A parsed tool call returned by the LLM.

id: str
function_name: str
arguments: str
args: Any
async run(*args, **kwargs)[source]

Execute the tool call. Only works when args is a Tool instance.

Extra positional and keyword arguments are forwarded to Tool.run(), allowing request-scoped state to be injected without closures.

Parameters:
Return type:

Any

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class derp.ai.models.ChatResponse[source]

Bases: BaseModel

Non-streaming chat completion response.

content: str
role: str
model: str
usage: Usage | None
finish_reason: str
tool_calls: list[ToolCall]
vercel_ai_json(*, message_id=None)[source]

Format as Vercel AI SDK data stream protocol events.

Parameters:

message_id (str | None)

Return type:

list[SSEEvent]

tanstack_ai_json(*, run_id=None, message_id=None)[source]

Format as TanStack AG-UI protocol events.

Parameters:

  • run_id (str | None)

  • message_id (str | None)

Return type:

list[SSEEvent]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class derp.ai.models.ToolEventType[source]

Bases: StrEnum

Type of tool lifecycle event emitted during agentic streaming.

INPUT_START = 'input_start'
INPUT_AVAILABLE = 'input_available'
OUTPUT_AVAILABLE = 'output_available'
__new__(value)
class derp.ai.models.ChatChunk[source]

Bases: BaseModel

A single chunk from a streaming chat completion.

delta: str
role: str
model: str | None
finish_reason: str | None
usage: Usage | None
tool_calls: list[ToolCall]
is_first: bool
is_last: bool
tool_event: ToolEventType | None
tool_call_id: str | None
tool_name: str | None
tool_input: dict[str, Any] | None
tool_output: Any
vercel_ai_json(*, message_id, stream_id='text-1')[source]

Format as Vercel AI SDK events.

Includes lifecycle events when is_first/is_last are set. When tool_event is set, emits tool lifecycle events instead.

Parameters:

  • message_id (str)

  • stream_id (str)

Return type:

list[SSEEvent]

tanstack_ai_json(*, message_id, run_id=None)[source]

Format as TanStack AG-UI events.

Includes lifecycle events when is_first/is_last are set. When tool_event is set, emits AG-UI tool call events instead.

Parameters:

  • message_id (str)

  • run_id (str | None)

Return type:

list[SSEEvent]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class derp.ai.models.CancelState[source]

Bases: StrEnum

Result of a job cancellation request.

CANCELLATION_REQUESTED = 'cancellation_requested'
ALREADY_COMPLETED = 'already_completed'
NOT_FOUND = 'not_found'
__new__(value)
class derp.ai.models.CancelResult[source]

Bases: BaseModel

Result of a cancel request with the job’s state at cancellation time.

state: CancelState
job_state: JobState
property is_cancelled: bool
property is_already_completed: bool
property is_not_found: bool
property job_failed: bool
property job_queued: bool
property job_in_progress: bool
property job_completed: bool
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class derp.ai.models.JobState[source]

Bases: StrEnum

State of an async AI job.

QUEUED = 'queued'
IN_PROGRESS = 'in_progress'
COMPLETED = 'completed'
FAILED = 'failed'
UNKNOWN = 'unknown'
__new__(value)
class derp.ai.models.JobStatus[source]

Bases: BaseModel

Status of an async AI job (e.g. fal image generation).

Wraps fal’s Queued/InProgress/Completed statuses into a single model.

state: JobState
position: int | None
logs: list[dict[str, Any]] | None
metrics: dict[str, Any] | None
error: str | None
error_type: str | None
property is_queued: bool
property is_in_progress: bool
property is_completed: bool
property is_failed: bool
property is_done: bool
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].