| 1 | """Helper function for pretty-printing content strings."""
|
| 2 |
|
| 3 | import json
|
| 4 |
|
| 5 |
|
| 6 | def _format_tool_call(args_str: str) -> str:
|
| 7 | """Format tool call arguments, extracting command if it's a bash call."""
|
| 8 | try:
|
| 9 | args = json.loads(args_str) if isinstance(args_str, str) else args_str
|
| 10 | if isinstance(args, dict) and "command" in args:
|
| 11 | return f"```\n{args['command']}\n```"
|
| 12 | except Exception:
|
| 13 | pass
|
| 14 | return f"```\n{args_str}\n```"
|
| 15 |
|
| 16 |
|
| 17 | def _format_observation(content: str) -> str | None:
|
| 18 | """Try to format an observation JSON as key-value pairs."""
|
| 19 | try:
|
| 20 | data = json.loads(content)
|
| 21 | if isinstance(data, dict) and "returncode" in data:
|
| 22 | lines = []
|
| 23 | for key, value in data.items():
|
| 24 | lines.append(f"<{key}>")
|
| 25 | lines.append(str(value))
|
| 26 | return "\n".join(lines)
|
| 27 | return content
|
| 28 | except Exception:
|
| 29 | return content
|
| 30 |
|
| 31 |
|
| 32 | def get_content_string(message: dict) -> str:
|
| 33 | """Extract text content from any message format for display.
|
| 34 |
|
| 35 | Handles:
|
| 36 | - Traditional chat: {"content": "text"}
|
| 37 | - Multimodal chat: {"content": [{"type": "text", "text": "..."}]}
|
| 38 | - Observation messages: {"content": "{\"returncode\": 0, \"output\": \"...\"}"}
|
| 39 | - Traditional tool calls: {"tool_calls": [{"function": {"name": "...", "arguments": "..."}}]}
|
| 40 | - Responses API: {"output": [{"type": "message", "content": [...]}]}
|
| 41 | """
|
| 42 | texts = []
|
| 43 |
|
| 44 | # Extract content (string or multimodal list)
|
| 45 | content = message.get("content")
|
| 46 | if isinstance(content, str):
|
| 47 | texts.append(_format_observation(content))
|
| 48 | elif isinstance(content, list):
|
| 49 | texts.append("\n".join(item.get("text", "") for item in content if isinstance(item, dict)))
|
| 50 |
|
| 51 | # Handle traditional tool_calls format (OpenAI/LiteLLM style)
|
| 52 | if tool_calls := message.get("tool_calls"):
|
| 53 | for tc in tool_calls:
|
| 54 | func = tc.get("function", {}) if isinstance(tc, dict) else getattr(tc, "function", None)
|
| 55 | if func:
|
| 56 | args = func.get("arguments", "{}") if isinstance(func, dict) else getattr(func, "arguments", "{}")
|
| 57 | texts.append(_format_tool_call(args))
|
| 58 |
|
| 59 | # Handle Responses API format (output array)
|
| 60 | if output := message.get("output"):
|
| 61 | if isinstance(output, str):
|
| 62 | texts.append(_format_observation(output))
|
| 63 | elif isinstance(output, list):
|
| 64 | for item in output:
|
| 65 | if not isinstance(item, dict):
|
| 66 | continue
|
| 67 | if item.get("type") == "message":
|
| 68 | for c in item.get("content", []):
|
| 69 | if isinstance(c, dict) and (text := c.get("text")):
|
| 70 | texts.append(text)
|
| 71 | elif item.get("type") == "function_call":
|
| 72 | texts.append(_format_tool_call(item.get("arguments", "{}")))
|
| 73 |
|
| 74 | return "\n\n".join(t for t in texts if t)
|
| 75 |
|