Python SDK Reference

pip install smartflow-sdk

pip install nest_asyncio

import asyncio
from smartflow import SmartflowClient

async def main():
    async with SmartflowClient("https://yourco.langsmart.app", api_key="sk-sf-...") as sf:  # gateway mode
        # Automatic caching, compliance scanning, multi-provider failover,
        # and complete audit logging on every call.
        response = await sf.chat("Explain quantum computing in simple terms")
        print(response)

asyncio.run(main())

from smartflow import SyncSmartflowClient

sf = SyncSmartflowClient()  # reads ~/.smartflow/config.yaml (gateway or direct)

response = sf.chat("What is machine learning?")
print(response)

stats = sf.get_cache_stats()
print(f"Cache hit rate: {stats.hit_rate:.1%}")
print(f"Tokens saved: {stats.tokens_saved:,}")

sf.close()

import nest_asyncio
nest_asyncio.apply()

from openai import OpenAI

# Before: client = OpenAI()
# After: through Smartflow — caching, compliance, logging all apply transparently

client = OpenAI(
    base_url="http://your-smartflow:7775/v1",
    api_key="sk-sf-your-virtual-key"
)

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}]
)

class SmartflowClient(
    base_url: Optional[str] = None, # Gateway URL or None (direct/config mode)
    api_key: Optional[str] = None,  # Virtual key (sk-sf-...)
    timeout: float = 30.0,          # Request timeout in seconds
    management_port: int = 7778,    # Management API port
    compliance_port: int = 7777,    # Compliance API port
    bridge_port: int = 3500,        # Hybrid bridge port
)

async with SmartflowClient("http://smartflow:7775", api_key="sk-sf-...") as sf:
    ...

# Or manual lifecycle
sf = SmartflowClient("http://smartflow:7775")
await sf._ensure_client()
# ... use sf ...
await sf.close()

async def chat(
    message: str,
    model: str = "gpt-4o",
    system_prompt: Optional[str] = None,
    temperature: float = 0.7,
    max_tokens: Optional[int] = None,
    **kwargs,
) -> str

async with SmartflowClient("http://smartflow:7775") as sf:

    # Simple
    response = await sf.chat("Explain Docker containers")

    # With options
    response = await sf.chat(
        message="Write a Python function to sort a list",
        model="gpt-4o",
        system_prompt="You are an expert Python developer. Write clean, documented code.",
        temperature=0.3,
        max_tokens=1000,
    )

async def chat_completions(
    messages: List[Dict[str, str]],
    model: str = "gpt-4o",
    temperature: float = 0.7,
    max_tokens: Optional[int] = None,
    stream: bool = False,
    **kwargs,
) -> AIResponse

response = await sf.chat_completions(
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What is REST API?"},
    ],
    model="gpt-4o",
)

print(response.content)
print(f"Tokens used: {response.usage.total_tokens}")
print(f"Cached: {response.cached}")  # True if served from MetaCache

async def stream_chat(
    message: str,
    model: str = "gpt-4o",
    system_prompt: Optional[str] = None,
    temperature: float = 0.7,
    max_tokens: Optional[int] = None,
    **kwargs,
) -> AsyncIterator[str]

async for chunk in sf.stream_chat("Tell me a story about a robot"):
    print(chunk, end="", flush=True)
print()

async def claude_message(
    message: str,
    model: str = "claude-sonnet-4-6",
    max_tokens: int = 1024,
    system: Optional[str] = None,
    anthropic_key: Optional[str] = None,
) -> str

response = await sf.claude_message(
    message="Analyze this code for security vulnerabilities",
    model="claude-sonnet-4-6",
    max_tokens=2000,
    system="You are a senior security engineer.",
)

async def embeddings(
    input: Union[str, List[str]],
    model: str = "text-embedding-3-small",
    encoding_format: str = "float",
    dimensions: Optional[int] = None,
    input_type: Optional[str] = None,
    **kwargs,
) -> Dict[str, Any]

# Single text
result = await sf.embeddings("Hello, world!")
vector = result["data"][0]["embedding"]

# Batch
result = await sf.embeddings([
    "First document",
    "Second document",
    "Third document",
])
vectors = [item["embedding"] for item in result["data"]]

# Cohere with input_type
result = await sf.embeddings(
    ["search query", "document text"],
    model="cohere/embed-english-v3.0",
    input_type="search_document",
)

# Reduce dimensions (OpenAI text-embedding-3+)
result = await sf.embeddings("Hello", model="text-embedding-3-large", dimensions=256)

async def image_generation(
    prompt: str,
    model: str = "dall-e-3",
    n: int = 1,
    size: str = "1024x1024",
    quality: Optional[str] = None,
    response_format: str = "url",
    style: Optional[str] = None,
    **kwargs,
) -> Dict[str, Any]

result = await sf.image_generation(
    "A futuristic city at sunrise",
    model="dall-e-3",
    size="1792x1024",
    quality="hd",
    style="vivid",
)
print(result["data"][0]["url"])

async def audio_transcription(
    file: Any,
    model: str = "whisper-1",
    language: Optional[str] = None,
    prompt: Optional[str] = None,
    response_format: str = "json",
    temperature: float = 0.0,
    filename: str = "audio.mp3",
    **kwargs,
) -> Dict[str, Any]

with open("recording.mp3", "rb") as f:
    result = await sf.audio_transcription(f, model="whisper-1")
print(result["text"])

# Groq (faster, free tier available)
with open("recording.mp3", "rb") as f:
    result = await sf.audio_transcription(f, model="groq/whisper-large-v3")

async def text_to_speech(
    input: str,
    model: str = "tts-1",
    voice: str = "alloy",
    response_format: str = "mp3",
    speed: float = 1.0,
    **kwargs,
) -> bytes

audio = await sf.text_to_speech("Hello, this is Smartflow.", voice="nova")
with open("output.mp3", "wb") as f:
    f.write(audio)

async def rerank(
    query: str,
    documents: List[str],
    model: str = "rerank-english-v3.0",
    top_n: Optional[int] = None,
    **kwargs,
) -> Dict[str, Any]

result = await sf.rerank(
    "What is the return policy?",
    ["We accept returns within 30 days.", "Contact support@example.com."],
    top_n=1,
)

async def list_models() -> List[Dict[str, Any]]

models = await sf.list_models()
for m in models:
    print(m["id"])

async def chatbot_query(query: str) -> Dict[str, Any]

result = await sf.chatbot_query("show me today's cache stats")
print(result["response"])

result = await sf.chatbot_query("which provider had the most errors this week?")
result = await sf.chatbot_query("what did we spend on OpenAI yesterday?")

# OpenAI — detected from gpt-*, o1-*, o3-*, chatgpt-*, whisper-*, dall-e-*
reply = await sf.chat("Hello", model="gpt-4o")
reply = await sf.chat("Hello", model="gpt-4o-mini")
reply = await sf.chat("Hello", model="o3-mini")

# Anthropic — detected from claude-*
reply = await sf.chat("Hello", model="claude-sonnet-4-6")
reply = await sf.chat("Hello", model="claude-3-opus-20240229")

# Google Gemini — detected from gemini-*
reply = await sf.chat("Hello", model="gemini-1.5-pro")
reply = await sf.chat("Hello", model="gemini-2.0-flash")

reply = await sf.chat("Hello", model="xai/grok-2-latest")
reply = await sf.chat("Hello", model="mistral/mistral-large-latest")
reply = await sf.chat("Hello", model="cohere/command-r-plus")
reply = await sf.chat("Hello", model="groq/llama-3.1-70b-versatile")
reply = await sf.chat("Hello", model="openrouter/meta-llama/llama-3.1-405b")
reply = await sf.chat("Hello", model="ollama/llama3.2")
reply = await sf.chat("Hello", model="azure/my-gpt4o-deployment")

# Force native Anthropic Messages API path
reply = await sf.claude_message("Hello", model="claude-sonnet-4-6")

async def intelligent_scan(
    content: str,
    user_id: Optional[str] = None,
    org_id: Optional[str] = None,
    context: Optional[str] = None,
) -> IntelligentScanResult

result = await sf.intelligent_scan(
    content="Please send payment to card 4111-1111-1111-1111",
)

print(f"Has violations: {result.has_violations}")      # True
print(f"Risk score: {result.risk_score:.2f}")          # 0.0 – 1.0
print(f"Action: {result.recommended_action}")          # Allow/AllowAndLog/Block/Review
print(f"Explanation: {result.explanation}")

for v in result.regex_violations:
    print(f"  - {v['violation_type']}: {v['severity']}")

result = await sf.intelligent_scan(
    content="Customer email: john.doe@example.com",
    user_id="support_agent_42",   # Track individual behavior
    org_id="acme_corporation",    # Compare against org baseline
    context="customer_support",   # Context for better detection
)

async def check_compliance(
    content: str,
    policy: str = "enterprise_standard",
) -> ComplianceResult

result = await sf.check_compliance("My SSN is 123-45-6789")
if result.has_violations:
    print(f"Risk: {result.risk_level}")
    print(f"PII: {result.pii_detected}")
    print(f"Safe text: {result.redacted_content}")

async def redact_pii(content: str) -> str

safe = await sf.redact_pii("Call me at 555-867-5309, email john@example.com")
# "Call me at [PHONE], email [EMAIL]"

async def submit_compliance_feedback(
    scan_id: str,
    is_false_positive: bool,
    user_id: Optional[str] = None,
    notes: Optional[str] = None,
) -> Dict[str, Any]

# Scan content, store the scan response dict to get the scan_id
response = await sf._post(
    f"{sf.compliance_url}/api/compliance/intelligent/scan",
    {"content": "Call me at 555-0100"}
)
scan_id = response.get("scan_id")

if scan_id:
    await sf.submit_compliance_feedback(
        scan_id=scan_id,
        is_false_positive=True,
        user_id="admin_user",
        notes="555-0100 is a known test number, not real PII",
    )

async def get_learning_summary() -> LearningSummary

summary = await sf.get_learning_summary()
print(f"Total users tracked: {summary.total_users}")
print(f"Users with complete baselines: {summary.users_learning_complete}")
print(f"Learning period: {summary.config_learning_days} days")

async def get_learning_status(user_id: str) -> LearningStatus

status = await sf.get_learning_status("user-alice")
print(f"Days tracked: {status.days_tracked}")
print(f"Progress: {status.progress_percent}%")
print(f"Complete: {status.learning_complete}")

async def get_ml_stats() -> MLStats

ml_stats = await sf.get_ml_stats()
print(f"Total patterns: {ml_stats.total_patterns}")
print(f"Learned patterns: {ml_stats.learned_patterns}")
print(f"Pattern categories: {ml_stats.patterns_by_category}")
print(f"Average confidence: {ml_stats.average_confidence:.2f}")

async def get_org_baseline(org_id: str) -> OrgBaseline

baseline = await sf.get_org_baseline("acme-corp")
print(f"Users: {baseline.user_count}")
print(f"Violation rate: {baseline.violation_rate:.2%}")
print(f"Top violations: {baseline.top_violation_types}")

await sf.register_mcp_server(
    server_id="github-tools",
    url="https://mcp.example.com/github",
    transport="streamable_http",
    description="GitHub issue and PR tooling",
)

servers = await sf.list_mcp_servers()
for s in servers:
    print(f"{s['server_id']:20s} {s['url']}")

probe = await sf.discover_mcp_tools("https://mcp.example.com/github")

for tool in probe["tools"]:
    print(f"{tool['name']}: {tool['description']}")

presets = await sf.list_mcp_connectors()
await sf.register_mcp_connector("github", scopes=["repo:read"])

result = await sf.call_mcp_tool(
    server_id="github-tools",
    name="create_issue",
    arguments={
        "repo": "my-org/my-repo",
        "title": "Bug: login fails on mobile",
        "body": "Steps to reproduce...",
    },
)

print(result["result"]["content"])

matches = await sf.search_mcp_tools("create a github issue", k=3)

for tool in matches:
    print(f"{tool['server_id']}.{tool['name']}: {tool['description']}")

catalog = await sf.get_mcp_catalog()              # every server
catalog = await sf.get_mcp_catalog("github-tools")  # one server

trust = await sf.get_mcp_trust("github-tools")
print(trust["score"], trust["signals"])

await sf.evaluate_mcp_trust("github-tools")

usage = await sf.get_mcp_usage()   # per-server call counts and cost totals

await sf.register_agent(
    agent_id="summarizer-agent",
    url="https://agents.internal/summarizer",
    description="Long-document summarisation",
)

for agent in await sf.list_agents():
    print(agent["agent_id"], agent["url"])

result = await sf.send_agent_task(
    agent_id="summarizer-agent",
    text="Summarise the Q4 earnings report.",
    trace_id="trace-abc-123",
)

print(result["result"]["parts"][0]["text"])

tasks = await sf.list_agent_tasks("summarizer-agent")
detail = await sf.get_agent_task(tasks[0]["id"])

card = await sf.get_agent_card("summarizer-agent")
print(card["name"], card["capabilities"])

cred = await sf.issue_agent_credential(
    agent_id="summarizer-agent",
    scopes=["documents:read", "summaries:write"],
    expires_in=86400,
)

print(cred["credential_id"])
print(cred["token"])   # give this to the agent; it is shown once

check = await sf.verify_agent_credential(token, required_scope="documents:read")

if not check["valid"]:
    raise PermissionError(check.get("reason", "invalid credential"))

jwks = await sf.get_aida_jwks()      # standard JWKS document
pub  = await sf.get_aida_pubkey()    # raw Ed25519 public key

for c in await sf.get_agent_credentials_for("summarizer-agent"):
    await sf.revoke_agent_credential(c["credential_id"])

await sf.create_policy({
    "name": "block-ssn-outbound",
    "description": "Reject prompts containing US SSNs",
    "rules": [{"match": "pii.ssn", "action": "block"}],
})

await sf.attach_policy({
    "policy_name": "block-ssn-outbound",
    "scope_type": "group",
    "scope_id": "contact-center",
})

for a in await sf.list_policy_attachments():
    print(f"{a['policy_name']} -> {a['scope_type']}:{a['scope_id']}")

resolution = await sf.resolve_policies({
    "user_id": "jane@acme.com",
    "groups": ["contact-center"],
    "model": "gpt-4o",
})

print(resolution["policies"])
print(resolution["active_frameworks"])

policies = await sf.list_builder_policies()
presets  = await sf.list_policy_presets()

await sf.assign_policy(
    policy_ids=["gdpr-baseline", "no-secrets-outbound"],
    subject_type="group",
    subject_id="eu-engineering",
)

for a in await sf.list_policy_assignments():
    print(a["subject_type"], a["subject_id"], a["policy_ids"])

job = await sf.generate_policies_from_document(
    text=open("data-handling-standard.md").read(),
    filename="data-handling-standard.md",
)

status = await sf.get_document_job(job["job_id"])
if status["state"] == "completed":
    drafts = await sf.get_document_job_results(job["job_id"])
    for d in drafts["policies"]:
        print(d["name"], "-", d["rationale"])

await sf.add_ai_inventory({
    "name": "Claims Triage Assistant",
    "purpose": "Route inbound claims by severity",
    "risk_tier": "high",
    "owner": "claims-ops@acme.com",
    "models": ["gpt-4o"],
})

for m in await sf.get_detected_models():
    print(f"undeclared: {m['model']} ({m['request_count']} requests)")

summary = await sf.get_conformity_summary()
print(f"{summary['covered']}/{summary['total']} articles covered")

for art in await sf.list_conformity_articles():
    if art["status"] != "covered":
        print(f"gap: Article {art['article']} - {art['title']}")

detail = await sf.get_conformity_article("13")   # transparency obligations

report = await sf.generate_examination_report({
    "framework": "eu_ai_act",
    "start": "2026-04-01",
    "end": "2026-06-30",
})

full = await sf.get_examination_report(report["report_id"])

integrity = await sf.verify_audit_chain()
print(integrity["valid"], integrity["entries_verified"])

entries = await sf.get_audit_logs(limit=500, user_id="jane@acme.com")

await sf.create_barrier({
    "name": "research-trading",
    "side_a": ["research"],
    "side_b": ["trading"],
    "mode": "block",
})

violations = await sf.list_barrier_violations()
attestation = await sf.get_barrier_attestation()

store = await sf.create_vector_store(name="policy-docs")

await sf.add_vector_store_file(
    store["id"],
    content=open("employee-handbook.md").read(),
    filename="employee-handbook.md",
)

hits = await sf.search_vector_store(store["id"], "parental leave", top_k=5)
for h in hits["results"]:
    print(f"{h['score']:.3f}  {h['text'][:80]}")

await sf.rag_ingest(content=open("q4-report.md").read(), filename="q4-report.md")

answer = await sf.rag_query("What drove the Q4 margin change?", top_k=4)
print(answer["answer"])
for src in answer["sources"]:
    print(" -", src["filename"])

await sf.set_sso_config({
    "provider": "entra",
    "issuer": "https://login.microsoftonline.com/<tenant-id>/v2.0",
    "client_id": "<client-id>",
    "client_secret": "<client-secret>",
    "group_claim": "groups",
})

status = await sf.get_sso_status()
print(status["enabled"], status["provider"])

for team in await sf.list_sso_teams():
    print(team["id"], team["name"])

stats = await sf.get_cache_stats()

print("=== CACHE PERFORMANCE ===")
print(f"Hit Rate: {stats.hit_rate:.1%}")
print(f"")
print(f"Layer Breakdown:")
print(f"  L1 (Memory):   {stats.l1_hits:,} hits")
print(f"  L2 (Semantic): {stats.l2_hits:,} hits")
print(f"  L3 (Exact):    {stats.l3_hits:,} hits")
print(f"")
print(f"Savings:")
print(f"  Tokens saved:  {stats.tokens_saved:,}")
print(f"  Cost saved:    ${stats.cost_saved_cents / 100:.2f}")

providers = await sf.get_provider_health()

print("=== PROVIDER STATUS ===")
for p in providers:
    print(f"{p.provider}")
    print(f"   Status:   {p.status}")
    print(f"   Latency:  {p.latency_ms:.0f}ms")
    print(f"   Success:  {p.success_rate:.1%}")
    print(f"   Requests: {p.requests_total:,}")

health = await sf.health_comprehensive()
print(health.status)          # "healthy" | "degraded" | "unhealthy"
print(health.uptime_seconds)
print(health.version)

status = await sf.health()
assert status["status"] == "ok"

async def get_logs(
    limit: int = 100,
    offset: int = 0,
    provider: Optional[str] = None,
    model: Optional[str] = None,
    days: int = 30,
) -> List[VASLog]

logs = await sf.get_logs(limit=50, provider="openai", days=7)

print("=== RECENT AI INTERACTIONS ===")
for log in logs:
    route = log.routing_strategy or "direct"
    cached_badge = f"[CACHE:{log.metacache.tokens_saved} tokens saved]" if log.metacache.hit else ""
    print(f"[{log.timestamp}] {log.provider}/{log.model} {cached_badge}")
    print(f"  Latency: {log.latency_ms}ms | Tokens: {log.tokens_used} | Routing: {route}")
    print(f"  Compliance: {log.compliance_status} | Conv: {log.conversation_id}")
    if log.compliance_violations:
        print(f"  Violations: {log.compliance_violations}")

page_size = 100
page = 0
while True:
    batch = await sf.get_logs(limit=page_size, offset=page * page_size)
    if not batch:
        break
    process(batch)
    page += 1

async def get_analytics(
    start_date: Optional[str] = None,
    end_date: Optional[str] = None,
) -> Dict[str, Any]

data = await sf.get_analytics(
    start_date="2026-02-01",
    end_date="2026-02-19",
)

status = await sf.get_routing_status()

async def force_provider(
    provider: str,
    duration_seconds: int = 300,
) -> Dict[str, Any]

# Force to OpenAI for 10 minutes during an Anthropic outage
await sf.force_provider("openai", duration_seconds=600)

from smartflow import SmartflowClient, SmartflowAgent

async with SmartflowClient("http://smartflow:7775") as sf:

    agent = SmartflowAgent(
        client=sf,
        name="TechSupport",
        model="gpt-4o",
        system_prompt="""You are a senior technical support engineer.
Guidelines:
- Be patient and thorough
- Ask clarifying questions when needed
- Provide step-by-step solutions
- Never ask for or repeat sensitive information""",
        temperature=0.7,
        compliance_policy="enterprise_standard",
        enable_compliance_scan=True,   # Auto-scan inputs and outputs
        user_id="support_session_123",
        org_id="tech_company",
    )

    # Conversation with full context memory
    print(await agent.chat("My application keeps crashing"))
    print(await agent.chat("It's a Python web app using Flask"))
    print(await agent.chat("Here's the error: MemoryError"))

    print(f"Messages exchanged: {agent.message_count}")
    agent.clear_history()

from smartflow import SmartflowClient, SmartflowWorkflow

async with SmartflowClient("http://smartflow:7775") as sf:

    workflow = SmartflowWorkflow(sf, name="ContentPipeline")

    workflow.add_step(
        name="analyze",
        action="chat",
        config={
            "prompt": "Analyze the tone and intent of this text: {input}",
            "model": "gpt-4o-mini",
        },
        next_steps=["compliance_check"],
    )

    workflow.add_step(
        name="compliance_check",
        action="compliance_check",
        config={"content": "{input}"},
        next_steps=["route"],
    )

    workflow.add_step(
        name="route",
        action="condition",
        config={
            "field": "output",
            "cases": {
                "positive": "enhance",
                "negative": "review",
                "neutral": "publish",
            },
        },
    )

    result = await workflow.execute({"input": "This product exceeded my expectations!"})

    print(f"Success: {result.success}")
    print(f"Path taken: {' -> '.join(result.steps_executed)}")
    print(f"Execution time: {result.execution_time_ms:.0f}ms")
    print(f"Total tokens: {result.total_tokens}")

from smartflow import SyncSmartflowClient

sf = SyncSmartflowClient("http://smartflow:7775", api_key="sk-sf-...")

reply      = sf.chat("Hello!")
emb        = sf.embeddings("Hello", model="text-embedding-3-small")
img        = sf.image_generation("A sunset", model="dall-e-3")
transcript = sf.audio_transcription(open("audio.mp3", "rb"), model="whisper-1")
audio      = sf.text_to_speech("Hello!", voice="nova")
ranked     = sf.rerank("What is the return policy?", ["doc1", "doc2"])
stats      = sf.get_cache_stats()
logs       = sf.get_logs(limit=20)

sf.close()

sf = SmartflowClient(
    base_url="http://smartflow:7775",   # Proxy endpoint
    api_key="sk-sf-...",                # Virtual key for authentication
    timeout=30.0,                       # Request timeout in seconds
    management_port=7778,               # Health, metrics, routing API
    compliance_port=7777,               # Compliance API
    bridge_port=3500,                   # Hybrid bridge (cross-instance logs)
)

import os
from smartflow import SmartflowClient

sf = SmartflowClient(
    base_url=os.environ["SMARTFLOW_URL"],
    api_key=os.environ.get("SMARTFLOW_API_KEY"),
)

from smartflow import (
    SmartflowClient,
    SmartflowError,
    ConnectionError,
    ComplianceError,
    RateLimitError,
    TimeoutError,
)
import asyncio

try:
    async with SmartflowClient("http://smartflow:7775") as sf:
        response = await sf.chat("Hello!")

except ConnectionError:
    print("Cannot connect to Smartflow proxy")

except ComplianceError as e:
    print(f"Blocked by compliance policy: {e}")

except RateLimitError:
    print("Rate limited — backing off")
    await asyncio.sleep(60)

except TimeoutError:
    print("Request timed out")

except SmartflowError as e:
    print(f"Smartflow error: {e}")

import asyncio
from smartflow import SmartflowClient, SmartflowAgent

class SecureCustomerSupportBot:
    """
    Customer support bot with built-in PII protection.

    - Automatic PII detection and blocking
    - Conversation memory
    - Audit trail for compliance
    - Behavioral analysis per customer
    """

    def __init__(self, smartflow_url: str):
        self.sf_url = smartflow_url

    async def handle_customer_session(self, customer_id: str, organization: str):
        async with SmartflowClient(self.sf_url) as sf:
            agent = SmartflowAgent(
                client=sf,
                name="SecureSupport",
                model="gpt-4o",
                system_prompt="""You are a helpful customer support agent for a financial services company.

CRITICAL RULES:
1. NEVER ask customers for full credit card numbers, SSNs, or passwords
2. If a customer shares sensitive info, acknowledge receipt but do not repeat it
3. For account verification, use last 4 digits only
4. Always offer secure channels for sensitive transactions""",
                compliance_policy="pci_dss_strict",
                enable_compliance_scan=True,
                user_id=f"customer_{customer_id}",
                org_id=organization,
            )

            print("SecureSupport: Hello! How can I help you today?")

            while True:
                user_input = input("Customer: ")
                if user_input.lower() == "quit":
                    break

                try:
                    response = await agent.chat(user_input)
                    print(f"SecureSupport: {response}")
                except Exception as e:
                    if "compliance" in str(e).lower():
                        print("SecureSupport: I noticed you shared some sensitive information.")
                        print("For your protection, please use our secure verification process.")
                    else:
                        print(f"Error: {e}")

async def main():
    bot = SecureCustomerSupportBot("http://smartflow:7775")
    await bot.handle_customer_session(customer_id="12345", organization="fintech_corp")

# asyncio.run(main())

import asyncio
from dataclasses import dataclass
from typing import List
from smartflow import SmartflowClient

@dataclass
class Product:
    id: str
    name: str
    category: str
    features: List[str]
    price: float

class ContentGenerationPipeline:
    """
    High-volume content generation with intelligent caching.

    Cost optimization:
    1. Similar products hit the semantic cache (60-80% savings)
    2. Smaller models for simple tasks
    3. Structured prompts to maximize cache hit potential
    """

    def __init__(self, smartflow_url: str):
        self.sf_url = smartflow_url

    async def generate_description(self, sf: SmartflowClient, product: Product) -> dict:
        # Structure prompt to maximize cache hits across similar products
        prompt = f"""Write a compelling product description.

Category: {product.category}
Product: {product.name}
Key Features: {', '.join(product.features)}
Price Point: ${product.price:.2f}

Requirements:
- 2-3 sentences
- Highlight key benefits
- Include call-to-action
- Professional tone"""

        response = await sf.chat_completions(
            messages=[{"role": "user", "content": prompt}],
            model="gpt-4o-mini",
            temperature=0.7,
        )

        return {
            "product_id": product.id,
            "description": response.content,
            "cached": response.cached,
            "tokens": response.usage.total_tokens,
        }

    async def process_catalog(self, products: List[Product]) -> dict:
        async with SmartflowClient(self.sf_url) as sf:
            initial_stats = await sf.get_cache_stats()
            results = []
            cached_count = 0
            total_tokens = 0

            for i, product in enumerate(products):
                result = await self.generate_description(sf, product)
                results.append(result)
                if result["cached"]:
                    cached_count += 1
                total_tokens += result["tokens"]

                if (i + 1) % 10 == 0:
                    print(f"Processed {i + 1}/{len(products)} products...")

            final_stats = await sf.get_cache_stats()
            tokens_saved = final_stats.tokens_saved - initial_stats.tokens_saved
            cost_saved = final_stats.cost_saved_cents - initial_stats.cost_saved_cents
            cache_hit_rate = cached_count / len(products) if products else 0

            return {
                "results": results,
                "summary": {
                    "total_products": len(products),
                    "cache_hit_rate": f"{cache_hit_rate:.1%}",
                    "tokens_used": total_tokens,
                    "tokens_saved": tokens_saved,
                    "cost_saved": f"${cost_saved / 100:.2f}",
                },
            }

async def main():
    pipeline = ContentGenerationPipeline("http://smartflow:7775")

    products = [
        Product("SKU001", "Wireless Bluetooth Headphones", "Electronics",
                ["Noise cancelling", "40hr battery", "Premium sound"], 149.99),
        Product("SKU002", "Wireless Earbuds Pro", "Electronics",
                ["Active noise cancelling", "36hr battery", "Hi-Fi audio"], 129.99),
        Product("SKU003", "Over-Ear Gaming Headset", "Electronics",
                ["7.1 surround", "Noise isolation", "RGB lighting"], 89.99),
    ]

    result = await pipeline.process_catalog(products)

    print(f"Products processed: {result['summary']['total_products']}")
    print(f"Cache hit rate:     {result['summary']['cache_hit_rate']}")
    print(f"Tokens saved:       {result['summary']['tokens_saved']:,}")
    print(f"Cost saved:         {result['summary']['cost_saved']}")

# asyncio.run(main())

import asyncio
from datetime import datetime
from smartflow import SmartflowClient, SmartflowAgent

class ResearchOrchestrator:
    """
    Multi-agent research system.

    Agents:
    1. Researcher — gathers and summarizes information
    2. Analyst    — identifies patterns and insights
    3. Writer     — produces polished executive report
    4. Editor     — reviews for accuracy and clarity (deep mode only)

    All interactions logged for full auditability.
    """

    def __init__(self, smartflow_url: str):
        self.sf_url = smartflow_url

    async def research_topic(self, topic: str, depth: str = "standard") -> dict:
        async with SmartflowClient(self.sf_url) as sf:
            timestamp = datetime.now().isoformat()

            researcher = SmartflowAgent(
                client=sf, name="Researcher", model="gpt-4o",
                system_prompt="""You are a thorough research analyst.
Provide structured findings covering: current state, recent developments,
key players, and challenges. Output as organized bullet points.""",
                user_id="research_system", org_id="analytics_dept",
            )

            print(f"Researcher: Investigating '{topic}'...")
            research_data = await researcher.chat(
                f"Research this topic: {topic}\n\n"
                f"Cover: current state and facts, recent developments, "
                f"key players, challenges and opportunities."
            )

            if depth == "quick":
                return {"topic": topic, "timestamp": timestamp,
                        "report": research_data, "agents_used": ["Researcher"]}

            analyst = SmartflowAgent(
                client=sf, name="Analyst", model="gpt-4o",
                system_prompt="""You are a strategic analyst.
Identify non-obvious patterns, provide data-driven insights,
make predictions, and highlight risks and opportunities.""",
                user_id="research_system", org_id="analytics_dept",
            )

            print("Analyst: Analyzing findings...")
            analysis = await analyst.chat(
                f"Analyze this research and provide strategic insights:\n\n"
                f"{research_data}\n\nFocus on hidden patterns, future implications, "
                f"and strategic recommendations."
            )

            writer = SmartflowAgent(
                client=sf, name="Writer", model="gpt-4o",
                system_prompt="""You are an expert business writer.
Synthesize research and analysis into a coherent executive narrative.
Lead with key findings and recommendations.""",
                user_id="research_system", org_id="analytics_dept",
            )

            print("Writer: Composing report...")
            draft_report = await writer.chat(
                f"Write an executive report from this research and analysis:\n\n"
                f"RESEARCH:\n{research_data}\n\nANALYSIS:\n{analysis}\n\n"
                f"Include: Executive Summary, Key Findings, "
                f"Strategic Analysis, Recommendations, Conclusion."
            )

            if depth == "standard":
                return {"topic": topic, "timestamp": timestamp,
                        "report": draft_report,
                        "agents_used": ["Researcher", "Analyst", "Writer"]}

            # Deep mode: add Editor
            editor = SmartflowAgent(
                client=sf, name="Editor", model="gpt-4o",
                system_prompt="""You are a senior business editor.
Review for accuracy and clarity, improve flow, ensure consistent tone,
fact-check against source research, polish for executive presentation.""",
                temperature=0.3,
                user_id="research_system", org_id="analytics_dept",
            )

            print("Editor: Polishing final report...")
            final_report = await editor.chat(
                f"Edit and polish this report:\n\n{draft_report}\n\n"
                f"Source research for fact-checking:\n{research_data}"
            )

            logs = await sf.get_logs(limit=10)

            return {
                "topic": topic,
                "timestamp": timestamp,
                "report": final_report,
                "agents_used": ["Researcher", "Analyst", "Writer", "Editor"],
                "audit_trail": [
                    {"timestamp": log.timestamp, "model": log.model,
                     "tokens": log.tokens_used, "cached": log.cached}
                    for log in logs
                ],
            }

async def main():
    orchestrator = ResearchOrchestrator("http://smartflow:7775")

    result = await orchestrator.research_topic(
        topic="The impact of AI agents on enterprise software development in 2026",
        depth="deep",
    )

    print("=" * 60)
    print(f"Topic: {result['topic']}")
    print(f"Generated: {result['timestamp']}")
    print(f"Agents: {' -> '.join(result['agents_used'])}")
    print("=" * 60)
    print(result["report"])

    print("\nAUDIT TRAIL:")
    for entry in result.get("audit_trail", []):
        cached = "[CACHED]" if entry["cached"] else ""
        print(f"  [{entry['timestamp']}] {entry['model']} — {entry['tokens']} tokens {cached}")

# asyncio.run(main())