Tools

def add(a: float, b: float) -> float:
    """Add two numbers together.

    Args:
        a: The first number
        b: The second number

    Returns:
        The sum of a and b
    """
    return a + b

def multiply(a: float, b: float) -> float:
    """Multiply two numbers.

    Args:
        a: The first number
        b: The second number

    Returns:
        The product of a and b
    """
    return a * b

from typing import Optional

def search_flights(
    destination: str,
    departure_date: str,
    flexible_days: int = 0,
    cabin_class: Optional[str] = None
) -> dict:
    """Search for available flights.

    Args:
        destination: The destination city (required)
        departure_date: The desired departure date (required)
        flexible_days: Number of flexible days for search. Defaults to 0.
        cabin_class: Preferred cabin class. Defaults to None.

    Returns:
        Dictionary with flight search results
    """
    results = {"destination": destination, "date": departure_date}

    if flexible_days > 0:
        results["flexible"] = True
        results["flex_days"] = flexible_days

    if cabin_class:
        results["cabin"] = cabin_class

    return results

import httpx

async def web_search(query: str, num_results: int = 5) -> list[dict]:
    """Search the web for information.

    Args:
        query: The search query
        num_results: Number of results to return (default: 5)

    Returns:
        List of search results with title, url, and snippet
    """
    print(f"Searching for: {query} (limit={num_results})")
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.search.example/search",
            params={"q": query, "limit": num_results}
        )
        results = response.json()["results"]
        print(f"Found {len(results)} results")
        return results

from connic import ToolFile


def export_invoice(invoice_id: str) -> ToolFile:
    """Generate an invoice PDF for the agent to inspect."""
    pdf = build_invoice_pdf(invoice_id)
    return ToolFile(
        mime_type="application/pdf",
        name=f"invoice-{invoice_id}.pdf",
        data=pdf,
    )

import os
import asyncpg
from typing import Any, Dict

async def fetch_user_account(user_id: str) -> Dict[str, Any]:
    """Fetch user account details from PostgreSQL by user ID.

    Args:
        user_id: The user's unique ID

    Returns:
        A dict with account details or a not-found marker
    """
    dsn = os.environ["POSTGRES_DSN"]
    conn = await asyncpg.connect(dsn)
    try:
        row = await conn.fetchrow(
            """
            SELECT id, email, status, plan, created_at
            FROM accounts
            WHERE id = $1
            """,
            user_id,
        )
        if not row:
            return {"found": False, "user_id": user_id}
        return {"found": True, "account": dict(row)}
    finally:
        await conn.close()

version: "1.0"

name: account-lookup
model: connic/gpt-5.6-terra
description: "Lookup account details during support workflows"
system_prompt: |
  Use the postgres.fetch_user_account tool to retrieve account info.
  If the account is missing, ask for the correct user ID.

tools:
  - postgres.fetch_user_account

version: "1.0"

name: assistant
model: connic/gpt-5.6-terra
description: "Assistant with calculator and search capabilities"
system_prompt: |
  You are a helpful assistant with access to tools.
  Use the calculator for math and web_search for current info.

tools:
  - calculator.add
  - calculator.multiply
  - search.web_search

  # Or replace the entries above with wildcards
  # - calculator.*               # all public functions in tools/calculator.py
  # - search.web_*               # functions starting with web_ in tools/search.py

from typing import Any, Dict

async def get_recent_orders(context: Dict[str, Any], limit: int = 5) -> list:
    """Fetch recent orders for the current user.

    Args:
        limit: Number of orders to return (default: 5)

    Returns:
        List of recent orders
    """
    # Read values set by middleware
    user_id = context.get("user_id")
    print(f"Fetching orders for user {user_id} (limit={limit})")

    # ... fetch orders from your database ...
    orders = [{"id": "ord-1", "user_id": user_id, "total": 49.99}]

    # Write values back to context
    context["orders_fetched"] = len(orders)
    print(f"Returning {len(orders)} orders")

    return orders

import os
import httpx

async def call_api(endpoint: str) -> dict:
    """Call an external API using stored credentials.

    Args:
        endpoint: The API endpoint to call

    Returns:
        API response as dictionary
    """
    # Access environment variables configured in Connic dashboard
    api_key = os.environ.get("EXTERNAL_API_KEY")
    base_url = os.environ.get("API_BASE_URL", "https://api.example.com")

    if not api_key:
        return {"error": "EXTERNAL_API_KEY not configured"}

    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{base_url}/{endpoint}",
            headers={"Authorization": f"Bearer {api_key}"}
        )
        return response.json()