Name: Copilot Python Sdk
Author: timsonner

搜索技能.../

Copilot Python Sdk | Skills Pool

import asyncio
from copilot import CopilotClient
from copilot.session import PermissionHandler

async def main():
    # Client automatically starts on enter and cleans up on exit
    async with CopilotClient() as client:
        # Create a session with automatic cleanup
        async with await client.create_session(model="gpt-5") as session:
            # Wait for response using session.idle event
            done = asyncio.Event()

            def on_event(event):
                if event.type.value == "assistant.message":
                    print(event.data.content)
                elif event.type.value == "session.idle":
                    done.set()

            session.on(on_event)

            # Send a message and wait for completion
            await session.send("What is 2+2?")
            await done.wait()

asyncio.run(main())

import asyncio
from copilot import CopilotClient
from copilot.session import PermissionHandler

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

    # Create a session (on_permission_request is required)
    session = await client.create_session(
        on_permission_request=PermissionHandler.approve_all,
        model="gpt-5",
    )

    done = asyncio.Event()

    def on_event(event):
        if event.type.value == "assistant.message":
            print(event.data.content)
        elif event.type.value == "session.idle":
            done.set()

    session.on(on_event)
    await session.send("What is 2+2?")
    await done.wait()

    # Clean up manually
    await session.disconnect()
    await client.stop()

asyncio.run(main())

from copilot import CopilotClient, SubprocessConfig
from copilot.session import PermissionHandler

async with CopilotClient() as client:
    async with await client.create_session(model="gpt-5") as session:
        def on_event(event):
            print(f"Event: {event.type}")

        session.on(on_event)
        await session.send("Hello!")

        # ... wait for events ...

from copilot import CopilotClient, ExternalServerConfig

# Connect to an existing CLI server
client = CopilotClient(ExternalServerConfig(url="localhost:3000"))

CopilotClient(
    config=None,        # SubprocessConfig | ExternalServerConfig | None
    *,
    auto_start=True,    # auto-start server on first use
    on_list_models=None, # custom handler for list_models()
)

# Get the session currently displayed in TUI (TUI+server mode only)
session_id = await client.get_foreground_session_id()

# Request TUI to display a specific session (TUI+server mode only)
await client.set_foreground_session_id("session-123")

# Subscribe to all lifecycle events
def on_lifecycle(event):
    print(f"{event.type}: {event.sessionId}")

unsubscribe = client.on(on_lifecycle)

# Subscribe to specific event type
unsubscribe = client.on("session.foreground", lambda e: print(f"Foreground: {e.sessionId}"))

# Later, to stop receiving events:
unsubscribe()

from pydantic import BaseModel, Field
from copilot import CopilotClient, define_tool

class LookupIssueParams(BaseModel):
    id: str = Field(description="Issue identifier")

@define_tool(description="Fetch issue details from our tracker")
async def lookup_issue(params: LookupIssueParams) -> str:
    issue = await fetch_issue(params.id)
    return issue.summary

async with await client.create_session(
    on_permission_request=PermissionHandler.approve_all,
    model="gpt-5",
    tools=[lookup_issue],
) as session:
    ...

from copilot import CopilotClient
from copilot.tools import Tool, ToolInvocation, ToolResult
from copilot.session import PermissionHandler

async def lookup_issue(invocation: ToolInvocation) -> ToolResult:
    issue_id = invocation.arguments["id"]
    issue = await fetch_issue(issue_id)
    return ToolResult(
        text_result_for_llm=issue.summary,
        result_type="success",
        session_log=f"Fetched issue {issue_id}",
    )

async with await client.create_session(
    on_permission_request=PermissionHandler.approve_all,
    model="gpt-5",
    tools=[
        Tool(
            name="lookup_issue",
            description="Fetch issue details from our tracker",
            parameters={
                "type": "object",
                "properties": {
                    "id": {"type": "string", "description": "Issue identifier"},
                },
                "required": ["id"],
            },
            handler=lookup_issue,
        )
    ],
) as session:
    ...

class EditFileParams(BaseModel):
    path: str = Field(description="File path")
    content: str = Field(description="New file content")

@define_tool(name="edit_file", description="Custom file editor with project-specific validation", overrides_built_in_tool=True)
async def edit_file(params: EditFileParams) -> str:
    # your logic

@define_tool(name="safe_lookup", description="A read-only lookup that needs no confirmation", skip_permission=True)
async def safe_lookup(params: LookupParams) -> str:
    # your logic

# File attachment — runtime reads from disk
await session.send(
    "What's in this image?",
    attachments=[
        {
            "type": "file",
            "path": "/path/to/image.jpg",
        }
    ],
)

# Blob attachment — provide base64 data directly
await session.send(
    "What's in this image?",
    attachments=[
        {
            "type": "blob",
            "data": base64_image_data,
            "mimeType": "image/png",
        }
    ],
)

await session.send("What does the most recent jpg in this directory portray?")

import asyncio
from copilot import CopilotClient
from copilot.session import PermissionHandler

async def main():
    async with CopilotClient() as client:
        async with await client.create_session(
            on_permission_request=PermissionHandler.approve_all,
            model="gpt-5",
            streaming=True,
        ) as session:
            # Use asyncio.Event to wait for completion
            done = asyncio.Event()

            def on_event(event):
                match event.type.value:
                    case "assistant.message_delta":
                        # Streaming message chunk - print incrementally
                        delta = event.data.delta_content or ""
                        print(delta, end="", flush=True)
                    case "assistant.reasoning_delta":
                        # Streaming reasoning chunk (if model supports reasoning)
                        delta = event.data.delta_content or ""
                        print(delta, end="", flush=True)
                    case "assistant.message":
                        # Final message - complete content
                        print("\n--- Final message ---")
                        print(event.data.content)
                    case "assistant.reasoning":
                        # Final reasoning content (if model supports reasoning)
                        print("--- Reasoning ---")
                        print(event.data.content)
                    case "session.idle":
                        # Session finished processing
                        done.set()

            session.on(on_event)
            await session.send("Tell me a short story")
            await done.wait()  # Wait for streaming to complete

asyncio.run(main())

# Default: infinite sessions enabled with default thresholds
async with await client.create_session(
    on_permission_request=PermissionHandler.approve_all,
    model="gpt-5",
) as session:
    # Access the workspace path for checkpoints and files
    print(session.workspace_path)
    # => ~/.copilot/session-state/{session_id}/

# Custom thresholds
async with await client.create_session(
    on_permission_request=PermissionHandler.approve_all,
    model="gpt-5",
    infinite_sessions={
        "enabled": True,
        "background_compaction_threshold": 0.80,  # Start compacting at 80% context usage
        "buffer_exhaustion_threshold": 0.95,  # Block at 95% until compaction completes
    },
) as session:
    ...

# Disable infinite sessions
async with await client.create_session(
    on_permission_request=PermissionHandler.approve_all,
    model="gpt-5",
    infinite_sessions={"enabled": False},
) as session:
    ...

async with await client.create_session(
    on_permission_request=PermissionHandler.approve_all,
    model="deepseek-coder-v2:16b",  # Required when using custom provider
    provider={
        "type": "openai",
        "base_url": "http://localhost:11434/v1",  # Ollama endpoint
        # api_key not required for Ollama
    },
) as session:
    await session.send("Hello!")

import os

async with await client.create_session(
    on_permission_request=PermissionHandler.approve_all,
    model="gpt-4",
    provider={
        "type": "openai",
        "base_url": "https://my-api.example.com/v1",
        "api_key": os.environ["MY_API_KEY"],
    },
) as session:
    ...

import os

async with await client.create_session(
    on_permission_request=PermissionHandler.approve_all,
    model="gpt-4",
    provider={
        "type": "azure",  # Must be "azure" for Azure endpoints, NOT "openai"
        "base_url": "https://my-resource.openai.azure.com",  # Just the host, no path
        "api_key": os.environ["AZURE_OPENAI_KEY"],
        "azure": {
            "api_version": "2024-10-21",
        },
    },
) as session:
    ...

from copilot import CopilotClient, SubprocessConfig

client = CopilotClient(SubprocessConfig(
    telemetry={
        "otlp_endpoint": "http://localhost:4318",
    },
))

from copilot import CopilotClient
from copilot.session import PermissionHandler

session = await client.create_session(
    on_permission_request=PermissionHandler.approve_all,
    model="gpt-5",
)

from copilot.session import PermissionRequestResult
from copilot.generated.session_events import PermissionRequest

def on_permission_request(request: PermissionRequest, invocation: dict) -> PermissionRequestResult:
    # request.kind — what type of operation is being requested:
    #   "shell"       — executing a shell command
    #   "write"       — writing or editing a file
    #   "read"        — reading a file
    #   "mcp"         — calling an MCP tool
    #   "custom-tool" — calling one of your registered tools
    #   "url"         — fetching a URL
    #   "memory"      — accessing or updating session/workspace memory
    #   "hook"        — invoking a registered hook
    # request.tool_call_id  — the tool call that triggered this request
    # request.tool_name     — name of the tool (for custom-tool / mcp)
    # request.file_name     — file being written (for write)
    # request.full_command_text — full shell command (for shell)

    if request.kind.value == "shell":
        # Deny shell commands
        return PermissionRequestResult(kind="denied-interactively-by-user")

    return PermissionRequestResult(kind="approved")

session = await client.create_session(
    on_permission_request=on_permission_request,
    model="gpt-5",
)

async def on_permission_request(request: PermissionRequest, invocation: dict) -> PermissionRequestResult:
    # Simulate an async approval check (e.g., prompting a user over a network)
    await asyncio.sleep(0)
    return PermissionRequestResult(kind="approved")

`kind` value	Meaning
`"approved"`	Allow the tool to run
`"denied-interactively-by-user"`	User explicitly denied the request
`"denied-no-approval-rule-and-could-not-request-from-user"`	No approval rule matched and user could not be asked (default when no kind is specified)
`"denied-by-rules"`	Denied by a policy rule
`"denied-by-content-exclusion-policy"`	Denied due to a content exclusion policy
`"no-result"`	Leave the request unanswered (not allowed for protocol v2 permission requests)

session = await client.resume_session(
    "session-id",
    on_permission_request=PermissionHandler.approve_all,
)

async def handle_user_input(request, invocation):
    # request["question"] - The question to ask
    # request.get("choices") - Optional list of choices for multiple choice
    # request.get("allowFreeform", True) - Whether freeform input is allowed

    print(f"Agent asks: {request['question']}")
    if request.get("choices"):
        print(f"Choices: {', '.join(request['choices'])}")

    # Return the user's response
    return {
        "answer": "User's answer here",
        "wasFreeform": True,  # Whether the answer was freeform (not from choices)
    }

async with await client.create_session(
    on_permission_request=PermissionHandler.approve_all,
    model="gpt-5",
    on_user_input_request=handle_user_input,
) as session:
    ...

async def on_pre_tool_use(input, invocation):
    print(f"About to run tool: {input['toolName']}")
    # Return permission decision and optionally modify args
    return {
        "permissionDecision": "allow",  # "allow", "deny", or "ask"
        "modifiedArgs": input.get("toolArgs"),  # Optionally modify tool arguments
        "additionalContext": "Extra context for the model",
    }

async def on_post_tool_use(input, invocation):
    print(f"Tool {input['toolName']} completed")
    return {
        "additionalContext": "Post-execution notes",
    }

async def on_user_prompt_submitted(input, invocation):
    print(f"User prompt: {input['prompt']}")
    return {
        "modifiedPrompt": input["prompt"],  # Optionally modify the prompt
    }

async def on_session_start(input, invocation):
    print(f"Session started from: {input['source']}")  # "startup", "resume", "new"
    return {
        "additionalContext": "Session initialization context",
    }

async def on_session_end(input, invocation):
    print(f"Session ended: {input['reason']}")

async def on_error_occurred(input, invocation):
    print(f"Error in {input['errorContext']}: {input['error']}")
    return {
        "errorHandling": "retry",  # "retry", "skip", or "abort"
    }

async with await client.create_session(
    on_permission_request=PermissionHandler.approve_all,
    model="gpt-5",
    hooks={
        "on_pre_tool_use": on_pre_tool_use,
        "on_post_tool_use": on_post_tool_use,
        "on_user_prompt_submitted": on_user_prompt_submitted,
        "on_session_start": on_session_start,
        "on_session_end": on_session_end,
        "on_error_occurred": on_error_occurred,
    },
) as session:
    ...

Copilot Python Sdk

Installation

Run the Sample

Quick Start

Copilot Python Sdk

Installation

Run the Sample

Quick Start

Manual Resource Management

Features

API Reference

CopilotClient

Tools

Overriding Built-in Tools

Skipping Permission Prompts

Image Support

Streaming

Infinite Sessions

Custom Providers

Telemetry

Permission Handling

Approve All (simplest)

Custom Permission Handler

Permission Result Kinds

Resuming Sessions

Per-Tool Skip Permission

User Input Requests

Session Hooks

Requirements

Openclaw Ghsa Maintainer

Gh Issues

Security Triage

Github

Author Contributions

Repository Setup