diff --git a/examples/basic/stream_function_call_args.py b/examples/basic/stream_function_call_args.py new file mode 100644 index 000000000..d3423a0eb --- /dev/null +++ b/examples/basic/stream_function_call_args.py @@ -0,0 +1,86 @@ +import asyncio +from typing import Any + +from openai.types.responses import ResponseFunctionCallArgumentsDeltaEvent + +from agents import Agent, Runner, function_tool + + +@function_tool +def write_file(filename: str, content: str) -> str: + """Write content to a file.""" + return f"File {filename} written successfully" + + +@function_tool +def create_config(project_name: str, version: str, dependencies: list[str]) -> str: + """Create a configuration file for a project.""" + return f"Config for {project_name} v{version} created" + + +async def main(): + """ + Demonstrates real-time streaming of function call arguments. + + Function arguments are streamed incrementally as they are generated, + providing immediate feedback during parameter generation. + """ + agent = Agent( + name="CodeGenerator", + instructions="You are a helpful coding assistant. Use the provided tools to create files and configurations.", + tools=[write_file, create_config], + ) + + print("šŸš€ Function Call Arguments Streaming Demo") + + result = Runner.run_streamed( + agent, + input="Create a Python web project called 'my-app' with FastAPI. Version 1.0.0, dependencies: fastapi, uvicorn" + ) + + # Track function calls for detailed output + function_calls: dict[Any, dict[str, Any]] = {} # call_id -> {name, arguments} + current_active_call_id = None + + async for event in result.stream_events(): + if event.type == "raw_response_event": + # Function call started + if event.data.type == "response.output_item.added": + if getattr(event.data.item, "type", None) == "function_call": + function_name = getattr(event.data.item, "name", "unknown") + call_id = getattr(event.data.item, "call_id", "unknown") + + function_calls[call_id] = { + 'name': function_name, + 'arguments': "" + } + current_active_call_id = call_id + print(f"\nšŸ“ž Function call streaming started: {function_name}()") + print("šŸ“ Arguments building...") + + # Real-time argument streaming + elif isinstance(event.data, ResponseFunctionCallArgumentsDeltaEvent): + if current_active_call_id and current_active_call_id in function_calls: + function_calls[current_active_call_id]['arguments'] += event.data.delta + print(f" + {event.data.delta}", end="", flush=True) + + # Function call completed + elif event.data.type == "response.output_item.done": + if hasattr(event.data.item, 'call_id'): + call_id = getattr(event.data.item, "call_id", "unknown") + if call_id in function_calls: + function_info = function_calls[call_id] + print(f"\nāœ… Function call streaming completed: {function_info['name']}") + print() + if current_active_call_id == call_id: + current_active_call_id = None + + print("Summary of all function calls:") + for call_id, info in function_calls.items(): + print(f" - #{call_id}: {info['name']}({info['arguments']})") + + print(f"\nResult: {result.final_output}") + + +if __name__ == "__main__": + asyncio.run(main())