import asyncio
import sys
from copilot import CopilotClient
from copilot.instruments import define_tool
from copilot.generated.session_events import SessionEventType
from pydantic import They’re fashions, Discipline
# Step 1: Outline customized instruments utilizing the @define_tool decorator.
class GetDataVisualizationParams(They’re fashions):
library_name: str = Discipline(description=“The title of the Python library to get information about”)
@define_tool(description=“Get details about a Python knowledge visualization library”)
async def get_library_info(params: GetDataVisualizationParams) -> dict:
“”“Customized device that gives details about knowledge visualization libraries.”“”
libraries = {
“matplotlib”: {
“title”: “Matplotlib”,
“use_case”: “Foundational plotting library for static, animated, and interactive visualizations”,
“set up”: “pip set up matplotlib”,
“recognition”: “Most generally used, foundation for a lot of different libraries”,
},
“seaborn”: {
“title”: “Seaborn”,
“use_case”: “Statistical knowledge visualization with engaging default kinds”,
“set up”: “pip set up seaborn”,
“recognition”: “Nice for exploratory knowledge evaluation”,
},
“plotly”: {
“title”: “Plotly”,
“use_case”: “Interactive, publication-quality graphs for dashboards”,
“set up”: “pip set up plotly”,
“recognition”: “Finest for web-based interactive visualizations”,
},
}
library = params.library_name.decrease()
if library in libraries:
return libraries[library]
return {“error”: f“Library ‘{library}’ not discovered. Attempt: matplotlib, seaborn, or plotly”}
async def most important():
# Step 2: Create and begin the Copilot shopper with an express CLI path.
# The SDK wants to search out the Copilot CLI, so specify the trail explicitly.
shopper = CopilotClient({
“cli_path”: “C:nvm4wnodejscopilot.cmd”, # Path to Copilot CLI
“log_level”: “debug”, # Allow debug logging for troubleshooting
})
print(“🚀 GitHub Copilot SDK Demo – Agentic Coding in Motion”)
print(“⏳ Beginning Copilot shopper (this may occasionally take a second)…n”)
await shopper.begin()
print(“=” * 60)
# Step 3: Create a session with customized configuration.
session = await shopper.create_session({
“mannequin”: “gpt-4.1”, # Select a mannequin
“streaming”: True, # Allow streaming responses
“instruments”: [get_library_info], # Register customized instruments
“system_message”: (
“You’re a useful technical assistant for knowledge scientists. “
“When requested about visualization libraries, use the get_library_info device “
“to supply correct data.”
),
})
print(f“Session created: {session.session_id}n”)
# Step 4: Arrange occasion handlers for streaming.
def handle_event(occasion):
if occasion.sort == SessionEventType.ASSISTANT_MESSAGE_DELTA:
# Stream the response because it arrives.
sys.stdout.write(occasion.knowledge.delta_content)
sys.stdout.flush()
elif occasion.sort == SessionEventType.TOOL_EXECUTION_START:
print(f“n🔧 Instrument known as: {occasion.knowledge.tool_name}”)
session.on(handle_event)
# Step 5: Ship a immediate and let the agent work.
print(“📝 Consumer: Listing three widespread Python libraries for knowledge visualization and their most important use case.n”)
print(“🤖 Assistant: “, finish=“”)
await session.send_and_wait({
“immediate”: (
“Listing three widespread Python libraries for knowledge visualization and their most important use case. “
“Use the get_library_info device to get correct details about every one.”
)
})
print(“nn” + “=” * 60)
# Step 6: Clear up.
await session.destroy()
await shopper.cease()
print(“✅ Session ended efficiently!”)
if __name__ == “__main__”:
asyncio.run(most important())
