GitHubCopilotSDKExample/interactive_assistant.py at main · Tarak-Chandra-Sarkar/GitHubCopilotSDKExample · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
import asyncio
import random
from rich.console import Console
from rich.panel import Panel
from rich.markdown import Markdown
from rich.prompt import Prompt
from rich.text import Text

from copilot import CopilotClient
from copilot.tools import define_tool
from copilot.session import PermissionHandler
from pydantic import BaseModel, Field

console = Console()

# ------------------ TOOL ------------------ #
class GetWeatherParams(BaseModel):
    city: str = Field(description="City name")

@define_tool(description="Get current weather for a specified city")
async def get_weather(params: GetWeatherParams) -> dict:
    conditions = ["sunny", "cloudy", "rainy", "overcast", "hazy"]
    return {
        "city": params.city,
        "temperature": f"{random.randint(5, 35)}°C",
        "condition": random.choice(conditions),
        "humidity": f"{random.randint(30, 90)}%"
    }

# ------------------ MAIN ------------------ #
async def main():
    console.print(Panel.fit(
        "[bold green]🌤️ Weather Assistant[/bold green]\n"
        "[dim]Type 'exit' to quit[/dim]\n\n"
        "Try:\n"
        "• What's the weather in Mumbai?\n"
        "• Compare weather in Kolkata and Chennai",
        border_style="green"
    ))

    async with CopilotClient() as client:
        session = await client.create_session(
            on_permission_request=PermissionHandler.approve_all,
            model="qwen3.5:0.8b",
            # model="gemma4:e2b",
            # model="gpt-4o",
            streaming=True,
            tools=[get_weather]
        )

        async with session:
            done = asyncio.Event()

            # Shared response buffer
            response_text = Text()

            # -------- EVENT HANDLER -------- #
            def on_event(event):
                nonlocal response_text

                match event.type.value:
                    case "assistant.message_delta":
                        delta = event.data.delta_content or ""
                        response_text.append(delta)

                    case "assistant.message":
                        # Final message handled after streaming
                        pass

                    case "session.idle":
                        done.set()

            session.on(on_event)

            # -------- CHAT LOOP -------- #
            while True:
                try:
                    user_input = Prompt.ask("\n[bold yellow]You[/bold yellow]")
                except EOFError:
                    break

                if user_input.lower() in ["exit", "quit"]:
                    console.print("[bold magenta]👋 Goodbye![/bold magenta]")
                    break

                if not user_input.strip():
                    console.print("[red]⚠️ Please enter something[/red]")
                    continue

                # -------- USER BUBBLE -------- #
                console.print(
                    Panel(
                        user_input,
                        title="[bold yellow]You[/bold yellow]",
                        border_style="yellow",
                        expand=False
                    )
                )

                # Reset state
                response_text = Text()
                done.clear()

                # -------- SEND & WAIT (silent streaming) -------- #
                try:
                    await session.send(user_input)
                    await done.wait()
                except Exception as e:
                    console.print(f"[red][Error][/red]: {e}")
                    continue

                # -------- ASSISTANT BUBBLE (FINAL CLEAN OUTPUT) -------- #
                console.print(
                    Panel(
                        Markdown(response_text.plain),
                        title="[bold blue]Assistant[/bold blue]",
                        border_style="blue"
                    )
                )

    console.print("[dim]Session closed.[/dim]")


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