feat(py): stream tool send_chunk and send_partial on generate

[huangjeff5]

During generate_stream, a tool can push mid-run updates two ways.
ctx.send_chunk(...) takes a Part or list of Parts. Each call arrives as a ModelResponseChunk with role='tool' and those parts.
ctx.send_partial(value) takes any value. Each call arrives as a tool-role chunk with one ToolResponsePart: this tool request's name and ref, metadata.partial True, and output set to the value you passed.
History and the next Agent turn keep only the completed tool return, including after a wire hop. tool.stream() yields send_chunk Parts only. Tool-role chunk.output stays empty.

# before — mid-tool updates during generate_stream were not stream events
@ai.tool(name='deploy')
async def deploy(_req: dict, ctx: ToolRunContext) -> str:
    ctx.send_chunk(Part(TextPart(text='svc-00042 is live')))
    ctx.send_partial({'step': 'uploading', 'percent': 50})
    return 'done'


stream = ai.generate_stream(prompt='deploy it', tools=[deploy])
async for chunk in stream.stream:
    print(chunk.role, chunk.text)
# only model chunks; the live text and the progress never appear

# after
@ai.tool(name='deploy')
async def deploy(_req: dict, ctx: ToolRunContext) -> str:
    ctx.send_chunk(Part(TextPart(text='svc-00042 is live')))
    ctx.send_partial({'step': 'uploading', 'percent': 50})
    return 'done'


stream = ai.generate_stream(prompt='deploy it', tools=[deploy])
async for chunk in stream.stream:
    if chunk.role != 'tool':
        continue
    for part in chunk.content:
        if (part.metadata or {}).get('partial'):
            print(part.tool_response.name, part.tool_response.output)
        else:
            print(chunk.text)
resp = await stream.response
# resp.messages has the tool return ("done"), not the mid-run updates

Decisions

• send_chunk takes a Part or list[Part]. Each call is one tool-role ModelResponseChunk with those parts. send_chunk is unstamped (no name or ref).
• send_partial(value) takes any value. Each call is one tool-role chunk with one ToolResponsePart: this tool request's name and ref, metadata.partial True, output is the value you passed. Two tools sending at once each stamp their own name and ref.
• History and the next Agent turn keep only the completed tool return, including after a wire hop. Mid-tool updates are not replayed on resume.
• tool.stream() yields send_chunk Parts only. send_partial does not appear there.
• Tool-role chunk.output stays empty. Read text or part.tool_response.output from the parts.
• await generate(...) / no listener: send_chunk validates then no-ops; send_partial no-ops.
• A listener error on a mid-tool update is dropped (the tool still returns). A listener error on the final tool-response chunk fails generate.
• Agent send_stream carries the same generate chunks. Walk the parts (metadata.partial / text) the same way.

[gemini-code-assist]

Code Review

This pull request implements support for streaming tool-role chunks via ToolRunContext.send_chunk during generate_stream. It updates the chunk accumulator to handle role changes properly, propagates the tool chunk streaming callback through the tool execution pipeline, and adds comprehensive tests. Feedback is provided on normalize_send_chunk_parts to optimize list validation by avoiding an unnecessary list copy and using typing.cast instead.

[gemini-code-assist]

For improved efficiency and code clarity, you can avoid creating a new list here. After validating the items, you can use typing.cast to inform the type checker about the list's contents. This avoids an unnecessary list copy and is a common pattern for this type of validation.

    if isinstance(chunk, list):
        for item in chunk:
            if not isinstance(item, Part):
                raise GenkitError(
                    status='INVALID_ARGUMENT',
                    message=f'send_chunk parts must be Part values, got {type(item).__name__}.',
                )
        return cast(list[Part], chunk)

[huangjeff5]

We keep the copy so send_chunk does not hand the caller's list through to generate. After this returns they can keep mutating that list; the streamed chunk stays what they passed.