main.py

"""
Copilot Agentic App - Python Quickstart
An interactive CLI application demonstrating custom tools with the Copilot SDK
"""

import asyncio
import sys
import os
from datetime import datetime
from dotenv import load_dotenv
from copilot import CopilotClient
from copilot.tools import define_tool
from copilot.generated.session_events import SessionEventType

# Load environment variables
load_dotenv()


# Example custom tool: Get current time
@define_tool(description="Get the current date and time")
async def get_time(params: dict) -> dict:
    """
    Returns the current date and time
    
    Args:
        params: Dictionary with optional 'timezone' parameter
    """
    now = datetime.now()
    return {
        "time": now.isoformat(),
        "timezone": params.get("timezone", "UTC"),
        "formatted": now.strftime("%Y-%m-%d %H:%M:%S")
    }


# Example custom tool: Simple calculator
@define_tool(description="Perform basic math calculations")
async def calculator(params: dict) -> dict:
    """
    Performs basic arithmetic operations
    
    Args:
        params: Dictionary with 'operation', 'a', and 'b' parameters
    """
    operation = params.get("operation")
    a = params.get("a")
    b = params.get("b")
    
    operations = {
        "add": lambda x, y: x + y,
        "subtract": lambda x, y: x - y,
        "multiply": lambda x, y: x * y,
        "divide": lambda x, y: x / y if y != 0 else float('nan')
    }
    
    if operation not in operations:
        raise ValueError(f"Unknown operation: {operation}")
    
    result = operations[operation](a, b)
    
    return {
        "result": result,
        "operation": operation,
        "inputs": {"a": a, "b": b}
    }


async def main():
    """Main application entry point"""
    print("🤖 Copilot Agentic App - Python")
    print("================================\n")
    
    # Initialize Copilot client
    client = CopilotClient()
    await client.start()
    
    # Create a session with custom tools
    session = await client.create_session({
        "model": os.getenv("COPILOT_MODEL", "gpt-4.1"),
        "streaming": os.getenv("COPILOT_STREAMING", "true").lower() != "false",
        "tools": [get_time, calculator],
    })
    
    print("✓ Copilot session created")
    print(f"✓ Model: {os.getenv('COPILOT_MODEL', 'gpt-4.1')}")
    print("✓ Custom tools loaded: get_time, calculator\n")
    
    # Handle streaming responses
    def handle_event(event):
        if event.type == SessionEventType.ASSISTANT_MESSAGE_DELTA:
            sys.stdout.write(event.data.delta_content)
            sys.stdout.flush()
        if event.type == SessionEventType.SESSION_IDLE:
            print("\n")
    
    session.on(handle_event)
    
    print("Type your questions or commands (type 'exit' to quit)")
    print("Try: 'What time is it?' or 'Calculate 15 + 27'\n")
    
    # Interactive CLI loop
    while True:
        try:
            user_input = input("You: ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n")
            break
        
        if user_input.lower() == "exit":
            print("\n👋 Goodbye!")
            break
        
        if not user_input:
            continue
        
        try:
            sys.stdout.write("Assistant: ")
            await session.send_and_wait({"prompt": user_input})
        except Exception as e:
            print(f"Error: {e}\n")
    
    # Cleanup
    await client.stop()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n👋 Goodbye!")
        sys.exit(0)
    except Exception as e:
        print(f"Fatal error: {e}", file=sys.stderr)
        sys.exit(1)