Getting started with GitHub Copilot part 2, streamable responses

#ai #githubcopilot #copilotsdk #python

I'm sure you've seen many AI apps where you sit tight for 30s or and you wonder if things are stuck? Not a great experience right? Yes, you're right, you deserve better, so how do we fix it?

Type your prompt> Tell me a joke
.
.
.
.
.
.
.
.
Why could I never find the atoms, cause they split..

Series on Copilot SDK

This series is about Copilot SDK and how you can leverage your existing GitHub Copilot license to integrate AI into your apps

Part 1 - install and your first app
Part 2 - streamable response, you're here

Addressing the problem

By streaming the response, the response now arrives in chunks, pieces that you can show as soon as they arrive. How can we do that though and how can GitHub Copilot SDK help out?

Well, there's two things you need to do:

Enable streaming. You need to set streaming to True when you call create_session.
Listen for events that contains a chunk. Specifically, you need to listen to ASSISTANT_MESSAGE_DELTA and print out the chunk.

# 1. Enable streaming
session = await client.create_session({
        "model": "gpt-4.1",
        "on_permission_request": PermissionHandler.approve_all,
        "streaming": True,
    })

print("Starting streamed response:")

# Listen for response chunks
def handle_event(event):
    if event.type == SessionEventType.ASSISTANT_MESSAGE_DELTA:
        // 2. Chunk arrived, print it
        sys.stdout.write(event.data.delta_content)
        sys.stdout.flush()
    if event.type == SessionEventType.SESSION_IDLE:
        print()  # New line when done

Here's what the full application looks like:

import asyncio
import sys
from copilot import CopilotClient, PermissionHandler
from copilot.generated.session_events import SessionEventType

async def main():
    client = CopilotClient()
    await client.start()

    session = await client.create_session({
        "model": "gpt-4.1",
        "on_permission_request": PermissionHandler.approve_all,
        "streaming": True,
    })

    print("Starting streamed response:")

    # Listen for response chunks
    def handle_event(event):
        if event.type == SessionEventType.ASSISTANT_MESSAGE_DELTA:
            sys.stdout.write(event.data.delta_content)
            sys.stdout.flush()
        if event.type == SessionEventType.SESSION_IDLE:
            print()  # New line when done

    session.on(handle_event)

    print("Sending prompt...")
    await session.send_and_wait({"prompt": "Tell me a short joke"})

    await client.stop()

if __name__ == "__main__":
    asyncio.run(main())