from shannon import ShannonClient # Initialize client (HTTP-only) client = ShannonClient( base_url="http://localhost:8080", api_key="your-api-key" # or use bearer_token ) # Submit a task handle = client.submit_task( "Analyze market trends for Q4 2024", session_id="my-session", ) print(f"Task submitted: {handle.task_id}") print(f"Workflow ID: {handle.workflow_id}") # Get status status = client.get_status(handle.task_id) print(f"Status: {status.status}") print(f"Progress: {status.progress:.1%}") # Cancel if needed # client.cancel(handle.task_id, reason="User requested") # Pause / resume controls # client.pause_task(handle.task_id, reason="User requested pause") # client.resume_task(handle.task_id, reason="User resumed") client.close()

from shannon import ShannonClient client = ShannonClient(base_url="http://localhost:8080", api_key="your-api-key") # Choose by tier (small|medium|large), or override model/provider explicitly handle = client.submit_task( "Summarize this document", model_tier="small", # model_override="gpt-5-nano-2025-08-07", # provider_override="openai", # mode="standard", # simple|standard|complex|supervisor ) final = client.wait(handle.task_id) print(final.result) client.close()

import asyncio from shannon import AsyncShannonClient async def main(): async with AsyncShannonClient( base_url="http://localhost:8080", api_key="your-api-key" ) as client: handle = await client.submit_task("What is 2+2?") final = await client.wait(handle.task_id) print(f"Result: {final.result}") asyncio.run(main())

import asyncio from shannon import AsyncShannonClient, EventType async def main(): async with AsyncShannonClient(base_url="http://localhost:8080") as client: h = await client.submit_task("Complex analysis") async for e in client.stream(h.workflow_id, types=[EventType.LLM_OUTPUT, EventType.WORKFLOW_COMPLETED]): if e.type == EventType.WORKFLOW_COMPLETED: break final = await client.wait(h.task_id) print(f"Result: {final.result}") asyncio.run(main())

from shannon import ShannonClient client = ShannonClient(base_url="http://localhost:8080", api_key="your-api-key") # Submit and get status handle = client.submit_task("Analyze quarterly revenue trends") status = client.wait(handle.task_id) # Access usage metadata (new in v0.3.0) print(f"Model used: {status.model_used}") print(f"Provider: {status.provider}") if status.usage: print(f"Total tokens: {status.usage.get('total_tokens')}") print(f"Prompt tokens: {status.usage.get('prompt_tokens')}") print(f"Completion tokens: {status.usage.get('completion_tokens')}") print(f"Cost: ${status.usage.get('cost_usd', 0):.6f}") # Access task metadata (citations, etc.) if status.metadata: print(f"Metadata: {status.metadata}")

tasks, total = client.list_tasks(limit=10) for t in tasks: tu = t.total_token_usage if tu: print(f"{t.task_id}: total={tu.total_tokens}, cost=${tu.cost_usd:.6f}")

# Get session with budget tracking session = client.get_session("my-session-id") print(f"Total cost: ${session.total_cost_usd:.4f}") print(f"Total tokens: {session.total_tokens_used}") print(f"Token budget: {session.token_budget}") print(f"Task count: {session.task_count}") # List sessions with metrics sessions, count = client.list_sessions(limit=10) for s in sessions: print(f"\nSession: {s.title or s.session_id}") print(f" Tasks: {s.successful_tasks} succeeded, {s.failed_tasks} failed") print(f" Success rate: {s.success_rate:.1%}") print(f" Total cost: ${s.total_cost_usd:.4f}") print(f" Avg cost/task: ${s.average_cost_per_task:.4f}") if s.token_budget: print(f" Budget: {s.budget_utilization:.1%} used ({s.budget_remaining} remaining)") if s.is_near_budget_limit: print(f" ⚠️ Near budget limit!")

from shannon import ShannonClient client = ShannonClient(base_url="http://localhost:8080", api_key="your-api-key") # Create tasks in a session handle = client.submit_task("Analyze Q4 revenue", session_id="quarterly-review") # Update session title for better organization client.update_session_title("quarterly-review", "Q4 2024 Financial Analysis") # Get session with title session = client.get_session("quarterly-review") print(f"Session: {session.title}")

sessions, _ = client.list_sessions(limit=20) for s in sessions: if s.is_active: print(f"✓ Active: {s.title or s.session_id}") if s.last_activity_at: print(f" Last used: {s.last_activity_at}") print(f" Latest task: {s.latest_task_query}") print(f" Status: {s.latest_task_status}")

session = client.get_session("my-session") if session.is_research_session: print(f"Research session using '{session.research_strategy}' strategy")

from shannon import ShannonClient client = ShannonClient(base_url="http://localhost:8080", api_key="your-api-key") # Create a daily research task at 9am UTC on weekdays result = client.create_schedule( name="Daily AI News Summary", cron_expression="0 9 * * 1-5", # Mon-Fri at 9am task_query="Summarize the latest developments in AI research", description="Daily automated research digest", timezone="UTC", task_context={ "force_research": "true", "research_strategy": "quick", }, max_budget_per_run_usd=0.50, timeout_seconds=300, ) print(f"Schedule created: {result['schedule_id']}") print(f"Next run: {result['next_run_at']}")

# List all active schedules schedules, total = client.list_schedules(status="ACTIVE") print(f"Found {total} active schedules") for s in schedules: print(f" {s.name}: {s.cron_expression} (next: {s.next_run_at})") # Get schedule details schedule = client.get_schedule("schedule-id") print(f"Schedule: {schedule.name}") print(f"Status: {schedule.status}") print(f"Runs: {schedule.total_runs} total, {schedule.successful_runs} succeeded, {schedule.failed_runs} failed") # Update schedule client.update_schedule( "schedule-id", cron_expression="0 10 * * 1-5", # Change to 10am max_budget_per_run_usd=1.00, ) # Pause/resume client.pause_schedule("schedule-id", reason="Holiday break") client.resume_schedule("schedule-id", reason="Back from holiday") # Delete client.delete_schedule("schedule-id")

# Get execution history for a schedule runs, total = client.get_schedule_runs("schedule-id", page=1, page_size=10) print(f"Last {len(runs)} runs (of {total} total):") for run in runs: status_icon = "✓" if run.status == "COMPLETED" else "✗" print(f" {status_icon} {run.triggered_at}: {run.status}") print(f" Tokens: {run.total_tokens}, Cost: ${run.total_cost_usd:.4f}") if run.error_message: print(f" Error: {run.error_message}")

from shannon import ShannonClient client = ShannonClient(base_url="http://localhost:8080") handle = client.submit_task( "Compare LangChain and AutoGen frameworks", context={ "research_strategy": "deep", "react_max_iterations": 6, "enable_verification": True, }, ) final = client.wait(handle.task_id) print(final.result) client.close()