import asyncio

from copilot import CopilotClient
from copilot.session_events import AssistantMessageData, SessionIdleData
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(
            on_permission_request=PermissionHandler.approve_all,
            model="gpt-5",
        ) as session:
            # Wait for response using session.idle event
            done = asyncio.Event()

            def on_event(event):
                match event.data:
                    case AssistantMessageData() as data:
                        print(data.content)
                    case SessionIdleData():
                        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_events import AssistantMessageData, SessionIdleData
from copilot.session import PermissionHandler


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

    # approve_all is only valid when managed settings are disabled.
    session = await client.create_session(
        on_permission_request=PermissionHandler.approve_all,
        model="gpt-5",
    )

    done = asyncio.Event()

    def on_event(event):
        match event.data:
            case AssistantMessageData() as data:
                print(data.content)
            case SessionIdleData():
                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
from copilot.session import PermissionHandler

async with CopilotClient() as client:
    async with await client.create_session(
        on_permission_request=PermissionHandler.approve_all,
        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, RuntimeConnection

# Connect to an existing CLI server
client = CopilotClient(connection=RuntimeConnection.for_uri("localhost:3000"))

CopilotClient()                                                   # spawn the bundled runtime with defaults
CopilotClient(connection=..., log_level="debug", github_token=..., ...)

conn = RuntimeConnection.for_stdio()
conn.env = {"MY_VAR": "value"}
client = CopilotClient(connection=conn)  # do NOT also pass env=... here

from copilot import CopilotClient, RuntimeConnection

client = CopilotClient(connection=RuntimeConnection.for_inprocess())
await client.start()
try:
    pong = await client.ping("hello")
    print(pong.message)
finally:
    await client.stop()

# 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.session_id}")


unsubscribe = client.on_lifecycle(on_lifecycle)

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

# 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:
    ...

tool = define_tool(
    "lookup_issue",
    description="Fetch issue details from our tracker",
    params_type=LookupIssueParams,
)

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

@define_tool(name="lookup_issue", description="Fetch issue details", defer="auto")
async def lookup_issue(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_events import (
    AssistantMessageData,
    AssistantMessageDeltaData,
    AssistantReasoningData,
    AssistantReasoningDeltaData,
    SessionIdleData,
)
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.data:
                    case AssistantMessageDeltaData() as data:
                        # Streaming message chunk - print incrementally
                        delta = data.delta_content or ""
                        print(delta, end="", flush=True)
                    case AssistantReasoningDeltaData() as data:
                        # Streaming reasoning chunk (if model supports reasoning)
                        delta = data.delta_content or ""
                        print(delta, end="", flush=True)
                    case AssistantMessageData() as data:
                        # Final message - complete content
                        print("\n--- Final message ---")
                        print(data.content)
                    case AssistantReasoningData() as data:
                        # Final reasoning content (if model supports reasoning)
                        print("--- Reasoning ---")
                        print(data.content)
                    case SessionIdleData():
                        # 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="gpt-5",
    memory={"enabled": True},
) 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:
    ...

async with await client.create_session(
    on_permission_request=PermissionHandler.approve_all,
    model="gpt-5",
    system_message={
        "mode": "append",
        "content": """
<workflow_rules>
- Always check for security vulnerabilities
- Suggest performance improvements when applicable
</workflow_rules>
""",
    },
) as session:
    ...

async with await client.create_session(
    on_permission_request=PermissionHandler.approve_all,
    model="gpt-5",
    system_message={
        "mode": "customize",
        "sections": {
            "tone": {
                "action": "replace",
                "content": "Respond in a warm, professional tone. Be thorough in explanations.",
            },
            "code_change_rules": {"action": "remove"},
            "guidelines": {"action": "append", "content": "\n* Always cite data sources"},
        },
        "content": "Focus on financial analysis and reporting.",
    },
) as session:
    ...

def redact_paths(content: str) -> str:
    return content.replace("/home/user", "/***")


async with await client.create_session(
    on_permission_request=PermissionHandler.approve_all,
    model="gpt-5",
    system_message={
        "mode": "customize",
        "sections": {
            "environment_context": {"action": redact_paths},
        },
    },
) as session:
    ...

async with await client.create_session(
    on_permission_request=PermissionHandler.approve_all,
    model="gpt-5",
    system_message={
        "mode": "replace",
        "content": "You are a helpful assistant.",
    },
) as session:
    ...

from copilot import CopilotClient

client = CopilotClient(
    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 import PermissionNoResult, PermissionRequest, PermissionRequestResult
from copilot.rpc import (
    PermissionDecisionApproveOnce,
    PermissionDecisionReject,
)
from copilot.session_events import PermissionRequestShell


def on_permission_request(request: PermissionRequest, invocation: dict) -> PermissionRequestResult:
    if getattr(request, "managed_approval_required", False) is True:
        return PermissionNoResult()

    # ``PermissionRequest`` is a discriminated union — pattern-match on
    # the variant class to access the per-kind fields.
    match request:
        case PermissionRequestShell(full_command_text=cmd):
            # Deny shell commands
            return PermissionDecisionReject(feedback=f"Shell denied: {cmd}")
        case _:
            return PermissionDecisionApproveOnce()


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

async def on_permission_request(
    request: PermissionRequest, invocation: dict
) -> PermissionRequestResult:
    if getattr(request, "managed_approval_required", False) is True:
        return PermissionNoResult()

    # Simulate an async approval check (e.g., prompting a user over a network)
    await asyncio.sleep(0)
    return PermissionDecisionApproveOnce()

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_post_tool_use_failure(input, invocation):
    # Fires when a tool's result was a failure. `on_post_tool_use` only fires
    # on success, so register this handler to observe failed tool calls. The
    # CLI extracts the failure message and passes it as the `error` field.
    print(f"Tool {input['toolName']} failed: {input['error']}")
    return {
        "additionalContext": f"Retry guidance for {input['toolName']}",
    }


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_post_tool_use_failure": on_post_tool_use_failure,
        "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:
    ...

from copilot.session import CommandDefinition, CommandContext, PermissionHandler


async def handle_deploy(ctx: CommandContext) -> None:
    print(f"Deploying with args: {ctx.args}")
    # ctx.session_id  — the session where the command was invoked
    # ctx.command      — full command text (e.g. "/deploy production")
    # ctx.command_name — command name without leading / (e.g. "deploy")
    # ctx.args         — raw argument string (e.g. "production")


async with await client.create_session(
    on_permission_request=PermissionHandler.approve_all,
    commands=[
        CommandDefinition(
            name="deploy",
            description="Deploy the app",
            handler=handle_deploy,
        ),
        CommandDefinition(
            name="rollback",
            description="Rollback to previous version",
            handler=lambda ctx: print("Rolling back..."),
        ),
    ],
) as session:
    ...

ui_caps = session.capabilities.get("ui", {})
if ui_caps.get("elicitation"):
    # Safe to call session.ui methods
    ...

ok = await session.ui.confirm("Deploy to production?")
if ok:
    print("Deploying...")

env = await session.ui.select("Choose environment:", ["staging", "production", "dev"])
if env:
    print(f"Selected: {env}")

name = await session.ui.input("Enter your name:")

# With options
email = await session.ui.input(
    "Enter email:",
    {
        "title": "Email Address",
        "description": "We'll use this for notifications",
        "format": "email",
    },
)

result = await session.ui.elicitation(
    {
        "message": "Configure deployment",
        "requestedSchema": {
            "type": "object",
            "properties": {
                "region": {"type": "string", "enum": ["us-east-1", "eu-west-1"]},
                "replicas": {"type": "number", "minimum": 1, "maximum": 10},
            },
            "required": ["region"],
        },
    }
)

if result["action"] == "accept":
    region = result["content"]["region"]
    replicas = result["content"].get("replicas", 1)

from copilot.session import ElicitationContext, ElicitationResult, PermissionHandler


async def handle_elicitation(
    context: ElicitationContext,
) -> ElicitationResult:
    # context["session_id"]           — the session ID
    # context["message"]              — what the server is asking
    # context.get("requestedSchema")  — optional JSON schema for form fields
    # context.get("mode")             — "form" or "url"

    print(f"Server asks: {context['message']}")

    # Return the user's response
    return {
        "action": "accept",  # or "decline" or "cancel"
        "content": {"answer": "yes"},
    }


async with await client.create_session(
    on_permission_request=PermissionHandler.approve_all,
    on_elicitation_request=handle_elicitation,
) as session:
    ...