Tool call data structure (OpenAI format).
| 70 | |
| 71 | |
| 72 | class ToolCallData(BaseModel): |
| 73 | """Tool call data structure (OpenAI format).""" |
| 74 | |
| 75 | id: str = Field(description="Tool call ID") |
| 76 | type: str = Field(default="function", description="Type of tool call") |
| 77 | function: FunctionCallData = Field(description="Function call data") |
| 78 | index: int = Field(default=0, description="Tool call index in batch") |
| 79 | |
| 80 | model_config = {"frozen": False} |
| 81 | |
| 82 | def to_dict(self) -> dict[str, Any]: |
| 83 | """Convert to dict for API.""" |
| 84 | return { |
| 85 | ToolCallField.ID: self.id, |
| 86 | ToolCallField.TYPE: self.type, |
| 87 | ToolCallField.FUNCTION: { |
| 88 | ToolCallField.NAME: self.function.name, |
| 89 | ToolCallField.ARGUMENTS: self.function.arguments, |
| 90 | }, |
| 91 | } |
| 92 | |
| 93 | @classmethod |
| 94 | def from_dict(cls, data: dict[str, Any]) -> "ToolCallData": |
| 95 | """Create from dict.""" |
| 96 | return cls( |
| 97 | id=data.get(ToolCallField.ID, ""), |
| 98 | type=data.get(ToolCallField.TYPE, "function"), |
| 99 | index=data.get(ToolCallField.INDEX, 0), |
| 100 | function=FunctionCallData( |
| 101 | name=data.get(ToolCallField.FUNCTION, {}).get(ToolCallField.NAME, ""), |
| 102 | arguments=data.get(ToolCallField.FUNCTION, {}).get( |
| 103 | ToolCallField.ARGUMENTS, "" |
| 104 | ), |
| 105 | ), |
| 106 | ) |
| 107 | |
| 108 | def merge_chunk(self, chunk: "ToolCallData") -> None: |
| 109 | """Merge data from a streaming chunk into this tool call. |
| 110 | |
| 111 | Args: |
| 112 | chunk: New chunk data to merge |
| 113 | """ |
| 114 | # Update function name if provided |
| 115 | if chunk.function.name: |
| 116 | self.function.name = chunk.function.name |
| 117 | |
| 118 | # Accumulate arguments (concatenate JSON strings) |
| 119 | if chunk.function.arguments: |
| 120 | self.function.arguments += chunk.function.arguments |
| 121 | |
| 122 | |
| 123 | class HistoryMessage(BaseModel): |
no outgoing calls