Module agents.llm.format_converters
Functions
def render_context_as_text(ctx: ChatContext) ‑> str-
Expand source code
def render_context_as_text(ctx: ChatContext) -> str: """Render a chat context as a compact plain-text transcript. Used to seed realtime providers that accept no structured history and only a system prompt (Ultravox). SYSTEM/DEVELOPER messages are skipped — the live system prompt already covers them. """ lines: list[str] = [] for item in ctx.items: if isinstance(item, ChatMessage): if item.role in (ChatRole.SYSTEM, ChatRole.DEVELOPER): continue if isinstance(item.content, str): text = item.content elif isinstance(item.content, list): text = " ".join(str(p) for p in item.content if isinstance(p, str)) else: text = "" if not text.strip(): continue speaker = "User" if item.role == ChatRole.USER else "Assistant" lines.append(f"{speaker}: {text.strip()}") elif isinstance(item, FunctionCall): lines.append(f"[tool call] {item.name}({item.arguments})") elif isinstance(item, FunctionCallOutput): tag = "tool error" if item.is_error else "tool result" lines.append(f"[{tag}] {item.name} -> {item.output}") return "\n".join(lines)Render a chat context as a compact plain-text transcript.
Used to seed realtime providers that accept no structured history and only a system prompt (Ultravox). SYSTEM/DEVELOPER messages are skipped — the live system prompt already covers them.
def to_anthropic_messages(ctx: ChatContext, *, caching: bool = False) ‑> tuple[list[dict], Optional[str]]-
Expand source code
def to_anthropic_messages(ctx: ChatContext, *, caching: bool = False) -> tuple[list[dict], Optional[str]]: """ Convert context to Anthropic messages format with role alternation enforced. Args: ctx: The ChatContext to convert. caching: If True, marks applicable messages with cache_control. Returns: tuple: (messages_list, system_content_or_none) """ anthropic_messages: list[dict] = [] system_content: str | None = None pending_tool_results: dict[str, FunctionCallOutput] = {} active_instructions, _ = ctx.active_config_at() has_config_updates = any(isinstance(it, AgentConfigUpdate) for it in ctx.items) for item in ctx.items: if isinstance(item, (AgentHandoff, AgentConfigUpdate)): continue if isinstance(item, ChatMessage): if item.role in (ChatRole.SYSTEM, ChatRole.DEVELOPER): if has_config_updates and active_instructions is not None: system_content = active_instructions elif isinstance(item.content, list): system_content = next( (str(p) for p in item.content if isinstance(p, str)), "" ) else: system_content = str(item.content) continue else: content = _format_content_anthropic(item.content) anthropic_messages.append({"role": item.role.value, "content": content}) elif isinstance(item, FunctionCall): tool_use_block = { "type": "tool_use", "id": item.call_id, "name": item.name, "input": ( json.loads(item.arguments) if isinstance(item.arguments, str) else item.arguments ), } if ( anthropic_messages and anthropic_messages[-1]["role"] == "assistant" and isinstance(anthropic_messages[-1]["content"], list) and any(p.get("type") == "tool_use" for p in anthropic_messages[-1]["content"]) ): anthropic_messages[-1]["content"].append(tool_use_block) else: anthropic_messages.append({"role": "assistant", "content": [tool_use_block]}) elif isinstance(item, FunctionCallOutput): pending_tool_results[item.call_id] = item final_messages: list[dict] = [] for msg in anthropic_messages: final_messages.append(msg) if isinstance(msg.get("content"), list) and any( part.get("type") == "tool_use" for part in msg["content"] ): tool_result_blocks = [] for part in msg["content"]: if part.get("type") != "tool_use": continue call_id = part["id"] if call_id in pending_tool_results: tool_result = pending_tool_results.pop(call_id) tool_result_blocks.append({ "type": "tool_result", "tool_use_id": call_id, "content": tool_result.output, "is_error": tool_result.is_error, }) if tool_result_blocks: final_messages.append({"role": "user", "content": tool_result_blocks}) merged: list[dict] = [] for msg in final_messages: if not merged or merged[-1]["role"] != msg["role"]: merged.append({"role": msg["role"], "content": msg["content"]}) else: prev_content = merged[-1]["content"] new_content = msg["content"] if isinstance(prev_content, str): prev_content = [{"type": "text", "text": prev_content}] if isinstance(new_content, str): new_content = [{"type": "text", "text": new_content}] merged[-1]["content"] = prev_content + new_content return merged, system_contentConvert context to Anthropic messages format with role alternation enforced.
Args
ctx- The ChatContext to convert.
caching- If True, marks applicable messages with cache_control.
Returns
tuple- (messages_list, system_content_or_none)
async def to_google_contents(ctx: ChatContext, *, thought_signatures: dict | None = None) ‑> tuple[list, Optional[str]]-
Expand source code
async def to_google_contents( ctx: ChatContext, *, thought_signatures: dict | None = None, ) -> tuple[list, Optional[str]]: """ Convert context to Google Gemini contents format. Args: ctx: The ChatContext to convert. thought_signatures: Dict mapping function names to thought signature bytes. Returns: tuple: (contents_list, system_instruction_or_none) Note: Requires ``google.genai.types`` to be available. """ try: from google.genai import types except ImportError: raise ImportError( "google-genai package is required for to_google_contents(). " "Install it with: pip install google-genai" ) thought_signatures = thought_signatures or {} contents = [] system_instruction = None active_instructions, _ = ctx.active_config_at() has_config_updates = any(isinstance(it, AgentConfigUpdate) for it in ctx.items) for item in ctx.items: if isinstance(item, (AgentHandoff, AgentConfigUpdate)): continue if isinstance(item, ChatMessage): if item.role in (ChatRole.SYSTEM, ChatRole.DEVELOPER): if has_config_updates and active_instructions is not None: system_instruction = active_instructions elif isinstance(item.content, list): system_instruction = next( (str(p) for p in item.content if isinstance(p, str)), "" ) else: system_instruction = str(item.content) continue parts = [] raw_content = item.content if isinstance(item.content, list) else [item.content] for part in raw_content: if part is None: continue if isinstance(part, str): parts.append(types.Part(text=part)) elif isinstance(part, ImageContent): data_url = part.to_data_url() if data_url.startswith("data:"): header, b64_data = data_url.split(",", 1) media_type = header.split(";")[0].split(":")[1] image_bytes = base64.b64decode(b64_data) parts.append( types.Part(inline_data=types.Blob(mime_type=media_type, data=image_bytes)) ) role = "model" if item.role == ChatRole.ASSISTANT else "user" contents.append(types.Content(role=role, parts=parts)) elif isinstance(item, FunctionCall): args = json.loads(item.arguments) if isinstance(item.arguments, str) else item.arguments function_call = types.FunctionCall(name=item.name, args=args) fc_part = types.Part(function_call=function_call) thought_sig_b64 = (item.metadata or {}).get("thought_signature") if thought_sig_b64: try: sig_bytes = base64.b64decode(thought_sig_b64) fc_part = types.Part(function_call=function_call, thought_signature=sig_bytes) except Exception: pass elif item.name in thought_signatures: fc_part = types.Part( function_call=function_call, thought_signature=thought_signatures[item.name], ) contents.append(types.Content(role="model", parts=[fc_part])) elif isinstance(item, FunctionCallOutput): function_response = types.FunctionResponse( name=item.name, response={"output": item.output} ) contents.append( types.Content(role="user", parts=[types.Part(function_response=function_response)]) ) return contents, system_instructionConvert context to Google Gemini contents format.
Args
ctx- The ChatContext to convert.
thought_signatures- Dict mapping function names to thought signature bytes.
Returns
tuple- (contents_list, system_instruction_or_none)
Note: Requires
google.genai.typesto be available. def to_openai_messages(ctx: ChatContext, *, reasoning_model: bool = False) ‑> list[dict]-
Expand source code
def to_openai_messages(ctx: ChatContext, *, reasoning_model: bool = False) -> list[dict]: """Convert context to OpenAI chat completion messages format. Skips structural items (AgentHandoff, AgentConfigUpdate). When the context contains AgentConfigUpdate items, the system/developer message emits the resolved active instructions instead of its literal content. Args: ctx: The ChatContext to convert. reasoning_model: If True, maps 'system' role to 'developer'. Returns: list[dict]: OpenAI-formatted messages list. """ active_instructions, _ = ctx.active_config_at() has_config_updates = any(isinstance(it, AgentConfigUpdate) for it in ctx.items) instruction_emitted = False openai_messages = [] i = 0 items = ctx.items while i < len(items): msg = items[i] if msg is None or isinstance(msg, (AgentHandoff, AgentConfigUpdate)): i += 1 continue if isinstance(msg, ChatMessage): role = msg.role.value if role == "developer" and not reasoning_model: role = "system" elif reasoning_model and role == "system": role = "developer" if msg.role in (ChatRole.SYSTEM, ChatRole.DEVELOPER): if instruction_emitted: i += 1 continue instruction_emitted = True if has_config_updates and active_instructions is not None: content: object = active_instructions else: content = _format_content_openai(msg.content) else: content = _format_content_openai(msg.content) openai_messages.append({ "role": role, "content": content, **({"name": msg.name} if hasattr(msg, "name") and getattr(msg, "name", None) else {}), }) i += 1 elif isinstance(msg, FunctionCall): tool_calls = [] while i < len(items) and isinstance(items[i], FunctionCall): fc = items[i] tool_calls.append({ "id": fc.call_id, "type": "function", "function": {"name": fc.name, "arguments": fc.arguments}, }) i += 1 openai_messages.append({ "role": "assistant", "content": None, "tool_calls": tool_calls, }) elif isinstance(msg, FunctionCallOutput): openai_messages.append({ "role": "tool", "tool_call_id": msg.call_id, "content": msg.output, }) i += 1 else: i += 1 return openai_messagesConvert context to OpenAI chat completion messages format.
Skips structural items (AgentHandoff, AgentConfigUpdate). When the context contains AgentConfigUpdate items, the system/developer message emits the resolved active instructions instead of its literal content.
Args
ctx- The ChatContext to convert.
reasoning_model- If True, maps 'system' role to 'developer'.
Returns
list[dict]- OpenAI-formatted messages list.