run_agent.py

# Tencent is pleased to support the open source community by making tRPC-Agent-Python available.
#
# Copyright (C) 2026 Tencent. All rights reserved.
#
# tRPC-Agent-Python is licensed under Apache-2.0.
"""
Streaming Tool Call Demo

This example demonstrates how to use StreamingFunctionTool to receive
real-time updates of tool call arguments as they are generated by the LLM.

Key features demonstrated:
1. StreamingFunctionTool with is_streaming=True property for enabling streaming
2. Automatic streaming detection via is_streaming property at runtime
3. Consuming streaming tool call events through Runner.run_async()
"""

import asyncio
import uuid

from dotenv import load_dotenv
from trpc_agent_sdk.models import TOOL_STREAMING_ARGS
from trpc_agent_sdk.runners import Runner
from trpc_agent_sdk.sessions import InMemorySessionService
from trpc_agent_sdk.types import Content
from trpc_agent_sdk.types import Part

load_dotenv()


async def run_streaming_tool_demo():
    """Run the streaming tool demo.

    This demo shows how streaming tool call arguments work:
    1. User asks the agent to create a file
    2. LLM generates the tool call arguments progressively
    3. Streaming events are yielded through Runner.run_async()
    4. Consume streaming events using event.is_streaming_tool_call()
    5. Finally, the complete tool call is executed
    """

    app_name = "streaming_tool_demo"

    from agent.agent import root_agent
    session_service = InMemorySessionService()
    runner = Runner(app_name=app_name, agent=root_agent, session_service=session_service)

    user_id = "demo_user"

    # Demo queries that will trigger write_file tool calls
    demo_queries = [
        "Please create a simple HTML page named index.html with a title and a paragraph.",
        # "Please write a Python script hello.py that implements a simple calculator.",
    ]

    for query in demo_queries:
        current_session_id = str(uuid.uuid4())

        await session_service.create_session(
            app_name=app_name,
            user_id=user_id,
            session_id=current_session_id,
        )

        print("=" * 60)
        print(f"🆔 Session ID: {current_session_id[:8]}...")
        print(f"📝 User: {query}")
        print("=" * 60)

        user_content = Content(parts=[Part.from_text(text=query)])

        print("\n🤖 Processing...\n")

        async for event in runner.run_async(user_id=user_id, session_id=current_session_id, new_message=user_content):
            if not event.content or not event.content.parts:
                continue

            if event.is_streaming_tool_call():
                # This is a streaming tool call event with partial arguments (delta mode)
                for part in event.content.parts:
                    if part.function_call:
                        args = part.function_call.args or {}
                        # Use delta mode - only show the delta content
                        delta = args.get(TOOL_STREAMING_ARGS, "")
                        if delta:
                            # Show a preview of the streaming delta
                            preview = delta[:60] + "..." if len(delta) > 60 else delta
                            print(f"⏳ [Streaming] {part.function_call.id}-{part.function_call.name}: {preview}")
                continue

            # Handle partial text responses (streaming text)
            if event.partial:
                for part in event.content.parts:
                    if part.text:
                        print(part.text, end="", flush=True)
                continue

            # Handle complete events
            for part in event.content.parts:
                if part.thought:
                    # Skip thinking content
                    continue
                if part.function_call:
                    # Complete tool call - this triggers actual execution
                    print(f"\n✅ [Tool Call Complete] {part.function_call.name}")
                    print(f"   Arguments: {part.function_call.args}")
                elif part.function_response:
                    # Tool execution result
                    print(f"\n📊 [Tool Result] {part.function_response.response}")
                elif part.text:
                    # Final text response
                    print(f"\n💬 {part.text}")

        print("\n" + "-" * 60 + "\n")


if __name__ == "__main__":
    print("""
╔══════════════════════════════════════════════════════════════╗
║           Streaming Tool Call Arguments Demo                 ║
╠══════════════════════════════════════════════════════════════╣
║  This demo shows real-time streaming of tool call arguments. ║
║  Watch as the LLM generates file content progressively!      ║
╚══════════════════════════════════════════════════════════════╝
    """)
    asyncio.run(run_streaming_tool_demo())