azad.ainetwork.types Module ¶

azad.ainetwork.types ¶

Attributes ¶

ANYJSON `module-attribute` ¶

ANYJSON = Dict[str, Any]

OneShotRequestUnion `module-attribute` ¶

OneShotRequestUnion = Union[OneShotTextRequest, OneShotImageRequest]

OneShotRequestAdapter `module-attribute` ¶

OneShotRequestAdapter = TypeAdapter(OneShotRequestUnion)

AINetworkEventUnion `module-attribute` ¶

AINetworkEventUnion = Union[NetworkAbortRequestAck, NetworkRequestAck, NetworkConnectionAttempt, AINetworkFirstToken, AINetworkConnectionEstablished, AINetworkConnectionEnded, NetworkConnectionFailed, NetworkConnectionInterrupted, AINetworkEventTextStart, AINetworkEventTextEnd, AINetworkEventTextChunk, AINetworkEventReasoningStart, AINetworkEventReasoningEnd, AIEventReasoningChunk, AINetworkEventToolName, AINetworkEventParameterStart, AINetworkEventParameterEnd, AINetworkEventParameterChunk, AINetworkEventToolReady, AINetworkEventParametersComplete, AINetworkEventContentComplete, NetworkImageGenerated, AINetworkEventCompressionTriggered, AINetworkEventCompressionNeeded, TaskMessageEvent, ServerToolUpdateEvent, AINetworkEventMultiToolStart, AINetworkEventMultiToolEnd, StepEndEvent, NetworkParseFailed]

Classes ¶

APIType ¶

Bases: str, Enum

Attributes ¶

TEXT_COMPLETION `class-attribute` `instance-attribute` ¶

TEXT_COMPLETION = 'text_completion'

IMAGE_GENERATION `class-attribute` `instance-attribute` ¶

IMAGE_GENERATION = 'image_generation'

BaseOneShotParams ¶

Bases: BaseModel

Base model for parameters, can be empty or hold common fields.

TextCompletionParams ¶

Bases: BaseOneShotParams

Parameters specific to text completion.

Attributes ¶

messages `class-attribute` `instance-attribute` ¶

messages: List[Message] = Field(..., description='List of messages for the completion request.')

tool_metadata `class-attribute` `instance-attribute` ¶

tool_metadata: Optional[List[ToolMetadata]] = Field(None, description='Optional tool schemas for the LLM.')

ImageGenerationParams ¶

Bases: BaseOneShotParams

Parameters specific to image generation, matching litellm.image_generation.

Attributes ¶

messages `class-attribute` `instance-attribute` ¶

messages: List[Message] = Field(description='Optional messages for the image generation request.')

prompt `class-attribute` `instance-attribute` ¶

prompt: str = Field(..., description='The text prompt for image generation.')

n `class-attribute` `instance-attribute` ¶

n: Optional[int] = Field(1, description='Number of images to generate (default 1).')

size `class-attribute` `instance-attribute` ¶

size: Optional[str] = Field(None, description="Image size (e.g., '1024x1024'). Varies by model.")

quality `class-attribute` `instance-attribute` ¶

quality: Optional[str] = Field(None, description="Image quality (e.g., 'hd', 'standard'). DALL-E 3 only.")

style `class-attribute` `instance-attribute` ¶

style: Optional[str] = Field(None, description="Image style (e.g., 'vivid', 'natural'). DALL-E 3 only.")

response_format `class-attribute` `instance-attribute` ¶

response_format: Optional[Literal['url', 'b64_json']] = Field('url', description="Response format (default 'url').")

user `class-attribute` `instance-attribute` ¶

user: Optional[str] = Field(None, description='Optional user identifier.')

StreamingContent ¶

Bases: BaseModel

A class for accumulating text in an 'append-only' fashion.

Functions ¶

append ¶

append(text: str) -> None

Source code in azad/ainetwork/types.py

def append(self, text: str) -> None:
    self._buffer.append(text)

new_content ¶

new_content() -> str

Source code in azad/ainetwork/types.py

def new_content(self) -> str:
    if self._last_index >= len(self._buffer):
        return ""
    new_chunks = self._buffer[self._last_index:]
    self._last_index = len(self._buffer)
    return "".join(new_chunks)

full_content ¶

full_content() -> str

Source code in azad/ainetwork/types.py

def full_content(self) -> str:
    return "".join(self._buffer)

set_completed ¶

set_completed() -> None

Source code in azad/ainetwork/types.py

def set_completed(self) -> None:
    self._completed = True

is_completed ¶

is_completed() -> bool

Source code in azad/ainetwork/types.py

def is_completed(self) -> bool:
    return self._completed

StreamingDict ¶

Bases: BaseModel

Maps parameter names to StreamingContent.

Functions ¶

get_or_create ¶

get_or_create(key: str) -> StreamingContent

Source code in azad/ainetwork/types.py

def get_or_create(self, key: str) -> StreamingContent:
    if key not in self._data:
        self._data[key] = StreamingContent()
    return self._data[key]

new_keys ¶

new_keys() -> List[str]

Source code in azad/ainetwork/types.py

def new_keys(self) -> List[str]:
    all_keys = set(self._data.keys())
    new = sorted(all_keys - self._known_keys)
    self._known_keys = all_keys
    return new

full_dict ¶

full_dict() -> Dict[str, str]

Source code in azad/ainetwork/types.py

def full_dict(self) -> Dict[str, str]:
    return {k: v.full_content() for k, v in self._data.items()}

set_completed ¶

set_completed(key: str) -> None

Source code in azad/ainetwork/types.py

def set_completed(self, key: str) -> None:
    if key in self._data:
        self._data[key].set_completed()

items ¶

items()

Source code in azad/ainetwork/types.py

def items(self):
    return self._data.items()

PromptResponse ¶

Bases: BaseModel

Class representing a parsed response from the AI.

Attributes ¶

tool_name `class-attribute` `instance-attribute` ¶

tool_name: str = Field(..., description='Name of the tool')

parameters `class-attribute` `instance-attribute` ¶

parameters: Dict[str, str] = Field(..., description='Tool parameters')

raw_response `class-attribute` `instance-attribute` ¶

raw_response: str = Field(..., description='Raw response content')

is_valid `class-attribute` `instance-attribute` ¶

is_valid: bool = Field(default=True, description='Whether the response is valid')

Functions ¶

create_invalid `classmethod` ¶

create_invalid(raw_response: str) -> 'PromptResponse'

Source code in azad/ainetwork/types.py

@classmethod
def create_invalid(cls, raw_response: str) -> 'PromptResponse':
    return cls(tool_name="", parameters={}, raw_response=raw_response, is_valid=False)

StreamingPromptResponse ¶

Bases: BaseModel

A streaming version of PromptResponse.

Attributes ¶

tool_name `class-attribute` `instance-attribute` ¶

tool_name: StreamingContent = Field(default_factory=StreamingContent, description='Tool name accumulator')

parameters `class-attribute` `instance-attribute` ¶

parameters: StreamingDict = Field(default_factory=StreamingDict, description='Parameters accumulator')

raw_response `class-attribute` `instance-attribute` ¶

raw_response: StreamingContent = Field(default_factory=StreamingContent, description='Raw response accumulator')

Functions ¶

set_completed ¶

set_completed() -> None

Source code in azad/ainetwork/types.py

def set_completed(self) -> None:
    self._completed = True
    self.tool_name.set_completed()
    for _, param_content in self.parameters.items():
        param_content.set_completed()

is_completed ¶

is_completed() -> bool

Source code in azad/ainetwork/types.py

def is_completed(self) -> bool:
    return self._completed

to_prompt_response ¶

to_prompt_response() -> PromptResponse

Source code in azad/ainetwork/types.py

def to_prompt_response(self) -> PromptResponse:
    if not self.is_completed():
        raise ValueError("Cannot convert incomplete streaming response")
    return PromptResponse(
        tool_name=self.tool_name.full_content(),
        parameters=self.parameters.full_dict(),
        raw_response=self.raw_response.full_content()
    )

BaseOneShotRequest ¶

Bases: BaseModel

Attributes ¶

api_type `class-attribute` `instance-attribute` ¶

api_type: str = Field(..., description='The type of API call to make.')

model_id `class-attribute` `instance-attribute` ¶

model_id: str = Field(..., description='The model ID to use for the request.')

model_api_key `class-attribute` `instance-attribute` ¶

model_api_key: str = Field(..., description='API key for the model.')

max_tokens `class-attribute` `instance-attribute` ¶

max_tokens: Optional[int] = Field(None, description='Maximum tokens for the request (0 or None means no limit).')

temperature `class-attribute` `instance-attribute` ¶

temperature: Optional[float] = Field(None, description='Temperature for the request.')

OneShotTextRequest ¶

Bases: BaseOneShotRequest

Attributes ¶

api_type `class-attribute` `instance-attribute` ¶

api_type: Literal[TEXT_COMPLETION] = TEXT_COMPLETION

params `class-attribute` `instance-attribute` ¶

params: TextCompletionParams = Field(..., description='Parameters for text completion.')

model_id `class-attribute` `instance-attribute` ¶

model_id: str = Field(..., description='The model ID to use for the request.')

model_api_key `class-attribute` `instance-attribute` ¶

model_api_key: str = Field(..., description='API key for the model.')

max_tokens `class-attribute` `instance-attribute` ¶

max_tokens: Optional[int] = Field(None, description='Maximum tokens for the request (0 or None means no limit).')

temperature `class-attribute` `instance-attribute` ¶

temperature: Optional[float] = Field(None, description='Temperature for the request.')

OneShotImageRequest ¶

Bases: BaseOneShotRequest

Attributes ¶

api_type `class-attribute` `instance-attribute` ¶

api_type: Literal[IMAGE_GENERATION] = IMAGE_GENERATION

params `class-attribute` `instance-attribute` ¶

params: ImageGenerationParams = Field(..., description='Parameters for image generation.')

model_id `class-attribute` `instance-attribute` ¶

model_id: str = Field(..., description='The model ID to use for the request.')

model_api_key `class-attribute` `instance-attribute` ¶

model_api_key: str = Field(..., description='API key for the model.')

max_tokens `class-attribute` `instance-attribute` ¶

max_tokens: Optional[int] = Field(None, description='Maximum tokens for the request (0 or None means no limit).')

temperature `class-attribute` `instance-attribute` ¶

temperature: Optional[float] = Field(None, description='Temperature for the request.')

AINetworkEvent ¶

Bases: BaseModel

Base class for all AI network events.

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

ErrorEventFields ¶

Bases: BaseModel

Attributes ¶

message `class-attribute` `instance-attribute` ¶

message: str = Field(..., description='Error message')

errName `class-attribute` `instance-attribute` ¶

errName: str = Field(..., description='Error name')

origErrName `class-attribute` `instance-attribute` ¶

origErrName: Optional[str] = Field(None, description='Original error name')

ContentEventFields ¶

Bases: BaseModel

Attributes ¶

streaming_response `class-attribute` `instance-attribute` ¶

streaming_response: Optional['StreamingPromptResponse'] = Field(default=None, description='Streaming response')

AINetworkLifecycleEvent ¶

Bases: AINetworkEvent

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkContentEvent ¶

Bases: AINetworkEvent

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkErrorEvent ¶

Bases: AINetworkEvent

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

NetworkConnectionAttempt ¶

Bases: AINetworkLifecycleEvent

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['network_attempt_connection'] = 'network_attempt_connection'

model_name `instance-attribute` ¶

model_name: str

assistant_id `instance-attribute` ¶

assistant_id: str

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkFirstToken ¶

Bases: AINetworkLifecycleEvent

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['network_first_token'] = 'network_first_token'

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkConnectionEstablished ¶

Bases: AINetworkLifecycleEvent

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['network_connected'] = 'network_connected'

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkConnectionEnded ¶

Bases: AINetworkLifecycleEvent

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['network_ended'] = 'network_ended'

cost `instance-attribute` ¶

cost: float

input_tokens `instance-attribute` ¶

input_tokens: int

output_tokens `instance-attribute` ¶

output_tokens: int

cache_read_tokens `class-attribute` `instance-attribute` ¶

cache_read_tokens: int | None = None

cache_write_tokens `class-attribute` `instance-attribute` ¶

cache_write_tokens: int | None = None

reasoning_tokens `class-attribute` `instance-attribute` ¶

reasoning_tokens: int | None = None

current_context_count `class-attribute` `instance-attribute` ¶

current_context_count: int | None = None

finish_reason `class-attribute` `instance-attribute` ¶

finish_reason: Literal['stop', 'length', 'content-filter', 'tool-calls', 'error', 'other', 'unknown'] = Field(..., description='Reason why the model stopped generating content')

generation_id `class-attribute` `instance-attribute` ¶

generation_id: str | None = None

raw_response_content `class-attribute` `instance-attribute` ¶

raw_response_content: str | None = None

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkEventTextChunk ¶

Bases: AINetworkContentEvent, ContentEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['text_chunk'] = 'text_chunk'

event: Text Chunk

content `class-attribute` `instance-attribute` ¶

content: str = Field(..., description='Chunk of text content')

content: Chunk of text content

streaming_response `class-attribute` `instance-attribute` ¶

streaming_response: Optional['StreamingPromptResponse'] = Field(default=None, description='Streaming response')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkEventTextStart ¶

Bases: AINetworkContentEvent, ContentEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['text_start'] = 'text_start'

streaming_response `class-attribute` `instance-attribute` ¶

streaming_response: Optional['StreamingPromptResponse'] = Field(default=None, description='Streaming response')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkEventTextEnd ¶

Bases: AINetworkContentEvent, ContentEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['text_end'] = 'text_end'

streaming_response `class-attribute` `instance-attribute` ¶

streaming_response: Optional['StreamingPromptResponse'] = Field(default=None, description='Streaming response')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkEventReasoningStart ¶

Bases: AINetworkContentEvent, ContentEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['reasoning_start'] = 'reasoning_start'

startedAt `class-attribute` `instance-attribute` ¶

startedAt: float = Field(..., description='Start time in milliseconds since epoch')

streaming_response `class-attribute` `instance-attribute` ¶

streaming_response: Optional['StreamingPromptResponse'] = Field(default=None, description='Streaming response')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkEventReasoningEnd ¶

Bases: AINetworkContentEvent, ContentEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['reasoning_end'] = 'reasoning_end'

endedAt `class-attribute` `instance-attribute` ¶

endedAt: float = Field(..., description='End time in milliseconds since epoch')

streaming_response `class-attribute` `instance-attribute` ¶

streaming_response: Optional['StreamingPromptResponse'] = Field(default=None, description='Streaming response')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AIEventReasoningChunk ¶

Bases: AINetworkContentEvent, ContentEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['reasoning_chunk'] = 'reasoning_chunk'

content `class-attribute` `instance-attribute` ¶

content: str = Field(..., description='Chunk of reasoning content')

signature `class-attribute` `instance-attribute` ¶

signature: str | None = Field(..., description='Signature of the reasoning chunk')

streaming_response `class-attribute` `instance-attribute` ¶

streaming_response: Optional['StreamingPromptResponse'] = Field(default=None, description='Streaming response')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkEventCompressionTriggered ¶

Bases: AINetworkContentEvent

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['compression_triggered'] = 'compression_triggered'

msg `instance-attribute` ¶

msg: CompressionMessage

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkEventCompressionNeeded ¶

Bases: AINetworkErrorEvent, ErrorEventFields

Event emitted when compression is needed but disabled in task configuration.

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['compression_needed'] = 'compression_needed'

token_count `class-attribute` `instance-attribute` ¶

token_count: int = Field(..., description='Current token count that exceeds the limit')

token_limit `class-attribute` `instance-attribute` ¶

token_limit: int = Field(..., description='Token limit from task configuration')

message `class-attribute` `instance-attribute` ¶

message: str = Field(..., description='Error message')

errName `class-attribute` `instance-attribute` ¶

errName: str = Field(..., description='Error name')

origErrName `class-attribute` `instance-attribute` ¶

origErrName: Optional[str] = Field(None, description='Original error name')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkEventToolName ¶

Bases: AINetworkContentEvent, ContentEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['tool_name'] = 'tool_name'

tool_name `class-attribute` `instance-attribute` ¶

tool_name: str = Field(description='Name of the tool')

tool_call_id `class-attribute` `instance-attribute` ¶

tool_call_id: str = Field(description='Tool call ID')

streaming_response `class-attribute` `instance-attribute` ¶

streaming_response: Optional['StreamingPromptResponse'] = Field(default=None, description='Streaming response')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkEventParameterStart ¶

Bases: AINetworkContentEvent, ContentEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['parameter_start'] = 'parameter_start'

parameter `class-attribute` `instance-attribute` ¶

parameter: str = Field(..., description='Parameter name starting')

streaming_response `class-attribute` `instance-attribute` ¶

streaming_response: Optional['StreamingPromptResponse'] = Field(default=None, description='Streaming response')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkEventParameterEnd ¶

Bases: AINetworkContentEvent, ContentEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['parameter_end'] = 'parameter_end'

parameter `class-attribute` `instance-attribute` ¶

parameter: str = Field(..., description='Parameter name ending')

streaming_response `class-attribute` `instance-attribute` ¶

streaming_response: Optional['StreamingPromptResponse'] = Field(default=None, description='Streaming response')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkEventParameterChunk ¶

Bases: AINetworkContentEvent, ContentEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['parameter_chunk'] = 'parameter_chunk'

parameter `class-attribute` `instance-attribute` ¶

parameter: str = Field(description='Parameter name')

content `class-attribute` `instance-attribute` ¶

content: str = Field(description='Parameter content chunk')

tool_call_id `class-attribute` `instance-attribute` ¶

tool_call_id: str = Field(description='Tool call ID')

streaming_response `class-attribute` `instance-attribute` ¶

streaming_response: Optional['StreamingPromptResponse'] = Field(default=None, description='Streaming response')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkEventToolReady ¶

Bases: AINetworkContentEvent, ContentEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['tool_ready'] = 'tool_ready'

tool_name `instance-attribute` ¶

tool_name: str

tool_call_id `instance-attribute` ¶

tool_call_id: str

args `instance-attribute` ¶

args: ANYJSON

streaming_response `class-attribute` `instance-attribute` ¶

streaming_response: Optional['StreamingPromptResponse'] = Field(default=None, description='Streaming response')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkEventParametersComplete ¶

Bases: AINetworkContentEvent, ContentEventFields

Attributes ¶

tool_call_id `instance-attribute` ¶

tool_call_id: str

event `class-attribute` `instance-attribute` ¶

event: Literal['parameters_complete'] = 'parameters_complete'

streaming_response `class-attribute` `instance-attribute` ¶

streaming_response: Optional['StreamingPromptResponse'] = Field(default=None, description='Streaming response')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkEventContentComplete ¶

Bases: AINetworkContentEvent, ContentEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['content_complete'] = 'content_complete'

streaming_response `class-attribute` `instance-attribute` ¶

streaming_response: Optional['StreamingPromptResponse'] = Field(default=None, description='Streaming response')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkEventExecutionApproved ¶

Bases: AINetworkContentEvent, ContentEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['execution_approved'] = 'execution_approved'

streaming_response `class-attribute` `instance-attribute` ¶

streaming_response: Optional['StreamingPromptResponse'] = Field(default=None, description='Streaming response')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkEventExecutionCanceled ¶

Bases: AINetworkContentEvent, ContentEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['execution_canceled'] = 'execution_canceled'

streaming_response `class-attribute` `instance-attribute` ¶

streaming_response: Optional['StreamingPromptResponse'] = Field(default=None, description='Streaming response')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkEventExecutionError ¶

Bases: AINetworkContentEvent, ContentEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['execution_error'] = 'execution_error'

streaming_response `class-attribute` `instance-attribute` ¶

streaming_response: Optional['StreamingPromptResponse'] = Field(default=None, description='Streaming response')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkEventExecutionAborted ¶

Bases: AINetworkContentEvent, ContentEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['execution_aborted'] = 'execution_aborted'

streaming_response `class-attribute` `instance-attribute` ¶

streaming_response: Optional['StreamingPromptResponse'] = Field(default=None, description='Streaming response')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

NetworkConnectionFailed ¶

Bases: AINetworkErrorEvent, ErrorEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['network_failed'] = 'network_failed'

message `class-attribute` `instance-attribute` ¶

message: str = Field(..., description='Error message')

errName `class-attribute` `instance-attribute` ¶

errName: str = Field(..., description='Error name')

origErrName `class-attribute` `instance-attribute` ¶

origErrName: Optional[str] = Field(None, description='Original error name')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

NetworkConnectionInterrupted ¶

Bases: AINetworkErrorEvent, ErrorEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['network_interrupted'] = 'network_interrupted'

message `class-attribute` `instance-attribute` ¶

message: str = Field(..., description='Error message')

errName `class-attribute` `instance-attribute` ¶

errName: str = Field(..., description='Error name')

origErrName `class-attribute` `instance-attribute` ¶

origErrName: Optional[str] = Field(None, description='Original error name')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

NetworkParseFailed ¶

Bases: AINetworkErrorEvent, ErrorEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['network_parse_failed'] = 'network_parse_failed'

message `class-attribute` `instance-attribute` ¶

message: str = Field(..., description='Error message')

errName `class-attribute` `instance-attribute` ¶

errName: str = Field(..., description='Error name')

origErrName `class-attribute` `instance-attribute` ¶

origErrName: Optional[str] = Field(None, description='Original error name')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

StepEndEvent ¶

Bases: AINetworkContentEvent

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['step_end'] = 'step_end'

new_messages `instance-attribute` ¶

new_messages: List[Message]

updated_state `instance-attribute` ¶

updated_state: TaskState

log `instance-attribute` ¶

log: str

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

TaskMessageEvent ¶

Bases: AINetworkContentEvent

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['task_message'] = 'task_message'

message `instance-attribute` ¶

message: Message

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

ServerToolUpdateEvent ¶

Bases: AINetworkContentEvent

Event carrying updates from a running server-side tool.

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['server_tool_update'] = 'server_tool_update'

server_tool_run_id `class-attribute` `instance-attribute` ¶

server_tool_run_id: str = Field(description='Unique ID for this specific tool run')

tool_name `class-attribute` `instance-attribute` ¶

tool_name: str = Field(description='Name of the tool being run')

step_number `class-attribute` `instance-attribute` ¶

step_number: Optional[int] = Field(None, description='Current step number, if applicable')

status `class-attribute` `instance-attribute` ¶

status: Literal['starting', 'running', 'progress', 'completed', 'error'] = Field(description='Current status of the tool execution')

data `class-attribute` `instance-attribute` ¶

data: ANYJSON = Field(default_factory=dict, description='Flexible payload for tool-specific data (screenshots, logs, results, errors)')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkEventMultiToolStart ¶

Bases: AINetworkContentEvent, ContentEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['multi_tool_start'] = 'multi_tool_start'

Event emitted when entering a multi-tool container.

streaming_response `class-attribute` `instance-attribute` ¶

streaming_response: Optional['StreamingPromptResponse'] = Field(default=None, description='Streaming response')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

AINetworkEventMultiToolEnd ¶

Bases: AINetworkContentEvent, ContentEventFields

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['multi_tool_end'] = 'multi_tool_end'

Event emitted when exiting a multi-tool container with the collected tools.

tool_count `class-attribute` `instance-attribute` ¶

tool_count: int = Field(0, description='Number of tools processed in the container')

streaming_response `class-attribute` `instance-attribute` ¶

streaming_response: Optional['StreamingPromptResponse'] = Field(default=None, description='Streaming response')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

ImageData ¶

Bases: BaseModel

Represents data for a single generated image from litellm response.

Attributes ¶

url `class-attribute` `instance-attribute` ¶

url: Optional[str] = Field(None, description='URL of the generated image, if available.')

b64_json `class-attribute` `instance-attribute` ¶

b64_json: Optional[str] = Field(None, description='Base64 encoded image data, if available.')

revised_prompt `class-attribute` `instance-attribute` ¶

revised_prompt: Optional[str] = Field(None, description='Optional prompt revision by the model.')

NetworkImageGenerated ¶

Bases: AINetworkContentEvent

Event indicating successful image generation.

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['image_generated'] = 'image_generated'

data `class-attribute` `instance-attribute` ¶

data: List[ImageData] = Field(..., description='List of generated image data.')

created `class-attribute` `instance-attribute` ¶

created: Optional[int] = Field(None, description='Timestamp of creation from the provider.')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

NetworkRequestAck ¶

Bases: AINetworkEvent

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['network_request_ack'] = 'network_request_ack'

request_id `class-attribute` `instance-attribute` ¶

request_id: str = Field(..., description='Unique identifier for the request.')

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}

NetworkAbortRequestAck ¶

Bases: AINetworkEvent

Attributes ¶

event `class-attribute` `instance-attribute` ¶

event: Literal['network_abort_request_ack'] = 'network_abort_request_ack'

Classes ¶

Config ¶

Attributes ¶

discriminator class-attribute instance-attribute ¶

discriminator = 'event'

json_encoders class-attribute instance-attribute ¶

json_encoders = {datetime: lambda dt: isoformat()}