feat(py): stream attributed progress from automatically executed tools

[huangjeff5]

Python tools can now call ToolRunContext.send_partial(...) to report structured, attributed progress while generate_stream is automatically executing them. Progress arrives as transient tool-role ToolResponsePart chunks carrying the active tool name and ref, and agent callers can consume the same updates through AgentChunk.tool_responses. Typed model output remains scoped to model-role chunks.

@ai.tool(name='deploy')
async def deploy(request: DeployRequest, ctx: ToolRunContext) -> DeployResult:
    ctx.send_partial({'stage': 'uploading', 'percent': 50})
    return DeployResult(url='https://example.run')

stream = ai.generate_stream(prompt='deploy it', tools=[deploy])
async for chunk in stream.stream:
    for response in chunk.tool_responses:
        if (response.metadata or {}).get('partial') is True:
            print(response.tool_response.name, response.tool_response.ref, response.tool_response.output)

Agent streams expose the same structured progress without adding it to conversation state:

turn = agent.chat().send_stream('deploy it')
async for chunk in turn.stream:
    for response in chunk.tool_responses:
        render_progress(response.tool_response.output)

Decisions:

• send_partial is the structured progress API. Its payload is wrapped in an attributed ToolResponsePart, so consumers can associate updates with the active call.
• send_chunk remains the raw streaming API for direct action execution only. Raw chunks are not injected into generate_stream, which keeps generated model and tool protocols distinct.
• Partial delivery is best-effort because a disconnected or failing stream sink must not change tool execution. The tool's final response remains authoritative.
• Progress is transient: it is excluded from model and agent history and is never replayed during resume or restart. This prevents stale status updates from becoming model input or durable conversation state.
• Concurrent tool progress is attributed by tool name and ref. Each tool's update order is preserved, while no cross-tool ordering guarantee is made because tools execute concurrently.
• AgentChunk.tool_responses exposes progress at the agent layer without requiring callers to decode raw transport chunks.
• Injecting raw ModelResponseChunk values from tools is out of scope. Tool progress has a structured wire shape rather than model-authored chunk semantics.

[gemini-code-assist]

Code Review

This pull request introduces support for streaming partial tool responses (progress updates) in Genkit. It adds tool_responses to AgentChunk and ModelResponseChunk, implements a mechanism to stream partial tool outputs via ToolRunContext.send_partial, and ensures that transient progress is skipped when stitching or persisting messages. The review feedback is highly constructive and identifies several key improvement opportunities: adding a defensive check in ModelResponseChunk.tool_responses to handle cases where self.content is None, binding the middleware-modified p.tool_request_part instead of the original trp or restart_trp when invoking on_partial in _generate.py, and storing references to background tasks created with asyncio.create_task in the test files to prevent premature garbage collection.

[gemini-code-assist]

If self.content is None, iterating over it will raise a TypeError. We should add a defensive check or default to an empty list to prevent potential runtime crashes when a chunk has no content.

Suggested change

	return [part.root for part in self.content if isinstance(part.root, ToolResponsePart)]
	return [part.root for part in (self.content or []) if isinstance(part.root, ToolResponsePart)]

[gemini-code-assist]

When invoking on_partial inside next_fn, we should bind p.tool_request_part instead of trp. If any wrap_tool middleware modifies the tool request part (e.g., updating metadata or input), using the original trp would pass stale information to the progress reporter.

Suggested change

	on_partial=partial(on_partial, trp) if on_partial is not None else None,
	on_partial=partial(on_partial, p.tool_request_part) if on_partial is not None else None,

[gemini-code-assist]

Similarly to the fresh tool execution path, we should bind p.tool_request_part instead of restart_trp when invoking on_partial inside next_fn for restarted tools. This ensures that any middleware-modified tool request part is correctly propagated to the progress reporter.

Suggested change

	on_partial=partial(on_partial, restart_trp) if on_partial is not None else None,
	on_partial=partial(on_partial, p.tool_request_part) if on_partial is not None else None,

[gemini-code-assist]

In Python's asyncio, tasks created with asyncio.create_task should have a reference kept to prevent them from being garbage collected mid-execution. We should store the task in a variable or a set to ensure it runs to completion reliably.

Suggested change

	asyncio.create_task(later())
	_task = asyncio.create_task(later())

[gemini-code-assist]

Similarly to the previous test, we should keep a reference to the background task created by asyncio.create_task to prevent it from being garbage collected prematurely.

Suggested change

	asyncio.create_task(later())
	_task = asyncio.create_task(later())