Skip to content

Instantly share code, notes, and snippets.

@compustar

compustar/copilot_sdk_04.ipynb

Created June 8, 2026 22:04
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save compustar/442835927fe23fea34b6f3d1d3704b00 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save compustar/442835927fe23fea34b6f3d1d3704b00 to your computer and use it in GitHub Desktop.
Download ZIP
copilot_sdk_04.ipynb
Display the source blob
Display the rendered blob
Raw
copilot_sdk_04.ipynb
{
"nbformat": 4,
"nbformat_minor": 0,
"metadata": {
"colab": {
"private_outputs": true,
"provenance": [],
"authorship_tag": "ABX9TyNMCvdNY//DJzzkBEmneVvM",
"include_colab_link": true
},
"kernelspec": {
"name": "python3",
"display_name": "Python 3"
},
"language_info": {
"name": "python"
}
},
"cells": [
{
"cell_type": "markdown",
"metadata": {
"id": "view-in-github",
"colab_type": "text"
},
"source": [
"<a href=\"https://colab.research.google.com/gist/compustar/442835927fe23fea34b6f3d1d3704b00/copilot_sdk_04.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"id": "sNF-MqHEChKC"
},
"outputs": [],
"source": [
"%%capture\n",
"!pip install github-copilot-sdk\n",
"!pip install nest-asyncio\n",
"\n",
"from IPython.display import HTML, display\n",
"\n",
"def set_css():\n",
" display(HTML('''\n",
" <style>\n",
" pre {\n",
" white-space: pre-wrap;\n",
" }\n",
" </style>\n",
" '''))\n",
"get_ipython().events.register('pre_run_cell', set_css)"
]
},
{
"cell_type": "code",
"source": [
"#@title Retrieve the API key\n",
"#@markdown\n",
"#@markdown Create new API key from AI Studio and setup the secrets in the sidebar\n",
"\n",
"from google.colab import userdata\n",
"llm_key_name = \"gemini-key\" #@param {type:\"string\"}\n",
"firecrawl_key_name = \"FIRECRAWL_API_KEY\" #@param {type:\"string\"}\n",
"api_key = userdata.get(llm_key_name)\n",
"firecrawl_api_key = userdata.get(firecrawl_key_name)\n",
"\n",
"#@markdown See https://ai.google.dev/gemini-api/docs/models for the list of valid models\n",
"model=\"gemini-3.1-flash-lite\" #@param {type:\"string\"}"
],
"metadata": {
"id": "wWJS5A38DCM0"
},
"execution_count": null,
"outputs": []
},
{
"cell_type": "code",
"source": [
"#@title Setup LLM Provider\n",
"#@markdown This configuration redirects the GitHub Copilot SDK to use the Gemini API via its OpenAI-compatible endpoint.\n",
"#@markdown This allows using the SDK without a GitHub Copilot subscription.\n",
"provider = {\n",
" \"type\": \"openai\",\n",
" \"base_url\": \"https://generativelanguage.googleapis.com/v1beta/openai/\",\n",
" \"wire_api\": \"completions\", # Use \"completions\" for older models\n",
" \"api_key\": api_key\n",
"}"
],
"metadata": {
"id": "hjVgGJ7GH8Mu"
},
"execution_count": null,
"outputs": []
},
{
"cell_type": "code",
"metadata": {
"id": "f3e40e04"
},
"source": [
"from typing import Any, Optional, List\n",
"from pydantic import BaseModel, Field\n",
"from copilot import define_tool\n",
"import httpx\n",
"from google.colab import userdata\n",
"\n",
"class FirecrawlSearchParams(BaseModel):\n",
" query: str = Field(..., description=\"The search query to find information on the web\")\n",
" limit: int = Field(default=5, description=\"Maximum number of results to return\")\n",
"\n",
"# Reverting to dict[str, Any] return type for better compatibility\n",
"async def fetch_firecrawl_results(params: FirecrawlSearchParams) -> dict[str, Any]:\n",
" endpoint = \"https://api.firecrawl.dev/v2/search\"\n",
"\n",
" headers = {\n",
" \"Authorization\": f\"Bearer {firecrawl_api_key}\",\n",
" \"Content-Type\": \"application/json\"\n",
" }\n",
"\n",
" payload = {\n",
" \"query\": params.query,\n",
" \"limit\": params.limit,\n",
" }\n",
"\n",
" async with httpx.AsyncClient() as client:\n",
" response = await client.post(endpoint, json=payload, headers=headers, timeout=30.0)\n",
" response.raise_for_status()\n",
" return response.json()\n",
"\n",
"@define_tool(\"web_search\", description=\"Search the web\")\n",
"async def firecrawl_search_tool(params: FirecrawlSearchParams) -> dict[str, Any]:\n",
" return await fetch_firecrawl_results(params)"
],
"execution_count": null,
"outputs": []
},
{
"cell_type": "code",
"metadata": {
"cellView": "form",
"id": "ce829364"
},
"source": [
"#@markdown **This cell is commented out and for testing purpose only**\n",
"# import asyncio\n",
"\n",
"# async def test_firecrawl():\n",
"# # Setting up the params for the search\n",
"# params = FirecrawlSearchParams(query=\"latest news on artificial intelligence\", limit=3)\n",
"\n",
"# print(f\"Searching for: {params.query}...\")\n",
"# # Call the logic function which now returns a dictionary\n",
"# result = await fetch_firecrawl_results(params)\n",
"\n",
"# # Accessing via dict keys instead of attributes\n",
"# if result.get(\"success\"):\n",
"# print(f\"Success: {result['success']}\")\n",
"# data = result.get(\"data\", [])\n",
"# for i, key in enumerate(data):\n",
"# for j, item in enumerate(data[key]):\n",
"# print(f\"\\n[{j+1}] Title: {item.get('title')}\")\n",
"# print(f\" URL: {item.get('url')}\")\n",
"# snippet = item.get('snippet')\n",
"# if snippet:\n",
"# print(f\" Snippet: {snippet[:100]}...\")\n",
"# else:\n",
"# print(f\"Search failed or returned no results: {result}\")\n",
"\n",
"\n",
"# # Run the test\n",
"# await test_firecrawl()"
],
"execution_count": null,
"outputs": []
},
{
"cell_type": "code",
"source": [
"import json\n",
"import asyncio\n",
"from copilot.generated.session_events import AssistantUsageData, AssistantMessageData, SystemMessageData, ToolExecutionStartData, SessionIdleData\n",
"\n",
"# Global queue for session events\n",
"event_queue = asyncio.Queue()\n",
"\n",
"#@title Event Handler\n",
"def handle_event(event):\n",
" # Put the event into the queue for the main loop to process\n",
" asyncio.get_event_loop().call_soon_threadsafe(event_queue.put_nowait, event)\n",
"\n",
"async def event_generator():\n",
" \"\"\"Asynchronous generator yielding dicts: {'title': ..., 'content': ...}\"\"\"\n",
" while True:\n",
" event = await event_queue.get()\n",
" try:\n",
" if isinstance(event.data, AssistantUsageData):\n",
" yield {\n",
" \"title\": \"Tokens used\",\n",
" \"content\": {\n",
" \"input\": event.data.input_tokens,\n",
" \"output\": event.data.output_tokens,\n",
" \"cache_read\": event.data.cache_read_tokens,\n",
" \"cache_write\": event.data.cache_write_tokens,\n",
" }\n",
" }\n",
" elif isinstance(event.data, SystemMessageData):\n",
" yield {\n",
" \"title\": \"System message\",\n",
" \"content\": json.dumps({\n",
" \"first_line\": event.data.content.split('\\n')[0],\n",
" \"content_length\": len(event.data.content)\n",
" }, indent=2)\n",
" }\n",
" elif isinstance(event.data, ToolExecutionStartData):\n",
" yield {\n",
" \"title\": \"Tool execution\",\n",
" \"content\": {\n",
" \"tool_name\": event.data.tool_name,\n",
" \"arguments\": event.data.arguments,\n",
" }\n",
" }\n",
" elif isinstance(event.data, AssistantMessageData):\n",
" if event.data.content:\n",
" yield {\n",
" \"title\": \"Assistant message\",\n",
" \"content\": event.data.content\n",
" }\n",
" elif isinstance(event.data, SessionIdleData):\n",
" break\n",
" finally:\n",
" event_queue.task_done()"
],
"metadata": {
"id": "uPcvGrDxE2e_"
},
"execution_count": null,
"outputs": []
},
{
"cell_type": "code",
"source": [
"import asyncio\n",
"import nest_asyncio\n",
"nest_asyncio.apply()\n",
"\n",
"from copilot import CopilotClient\n",
"from copilot.session import PermissionHandler, SystemMessageReplaceConfig\n",
"\n",
"async def main():\n",
" question = \"what is Qwen 3.7?\" #@param {type:\"string\"}\n",
" async with CopilotClient() as client:\n",
" async with await client.create_session(\n",
" on_permission_request=PermissionHandler.approve_all,\n",
" model=model,\n",
" provider=provider,\n",
" system_message=SystemMessageReplaceConfig(\n",
" mode=\"replace\",\n",
" content=\"You are a helpful assistant. Use web_search for queries.\"\n",
" ),\n",
" tools=[firecrawl_search_tool],\n",
" available_tools=['web_search']\n",
" ) as session:\n",
" session.on(handle_event)\n",
"\n",
" # Send the question without blocking\n",
" send_task = asyncio.create_task(session.send(question))\n",
"\n",
" # Simple one-liner loop processing the dict\n",
" async for info in event_generator():\n",
" print(f\"{info['title']}: {info['content']}\")\n",
"\n",
" await send_task\n",
"\n",
"asyncio.run(main())"
],
"metadata": {
"id": "0wWwRlcECp8l"
},
"execution_count": null,
"outputs": []
},
{
"cell_type": "code",
"source": [],
"metadata": {
"id": "A7hnuQ56Cu4h"
},
"execution_count": null,
"outputs": []
}
]
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment