Smartflow Python SDK

The official Python SDK for Smartflow - the enterprise AI orchestration, caching, compliance, and governance platform.

Features

• 🚀 Simple API - Chat, embeddings, images, audio, and completions with one line of code
• 💰 60-80% Cost Savings - 3-layer semantic caching (L1/L2/L3)
• 🛡️ ML-Powered Compliance - Intelligent PII detection with adaptive learning
• 🔄 Automatic Failover - Multi-provider routing with intelligent fallback
• 📊 Full Audit Trail - VAS logs + tamper-evident audit chain for every AI interaction
• 🔌 MCP Governance - Register, discover, trust-check, and invoke MCP tool servers
• 🤝 Agent Registry (A2A + AIDA) - Register agents, invoke them, and issue Ed25519 "virtual ID" credentials
• ⚖️ Policy Engine - Guardrail policies, attachments, and GUI-built policy assignments by user/group/region
• 📚 Governance Reporting - AI inventory, EU AI Act conformity, and examination/evidence reports
• 🤖 Agent Builder - Create AI agents with built-in compliance
• 📈 Workflow Orchestration - Chain AI operations with branching and error handling

What's new in 0.4.0

Dual-mode operation — the SDK now works with or without a Smartflow gateway. Plus MCP server registry + governed call_mcp_tool(), the A2A agent registry, AIDA cryptographic agent credentials, guardrail policies/attachments, the Policy Perfect builder (port 7782), governance/audit reporting, vector stores + RAG, SSO config, and the media methods (embeddings, image_generation, audio_transcription, text_to_speech, stream_chat, rerank). See CHANGELOG.md for the full list.

Dual-Mode Operation

Start building immediately without any infrastructure, then point at a gateway later — no code changes.

	Gateway Mode	Direct Mode
Chat / streaming / embeddings	✓	✓
Multi-provider routing	✓ 37+ providers	✓ OpenAI / Anthropic / Gemini / Ollama / local
Semantic cache, compliance, VAS audit, MCP/A2A, policies	✓	✗ raises DirectModeError
Required dependencies	httpx only	httpx only

Mode is resolved automatically, in priority order:

from smartflow import SmartflowClient

# 1. Explicit URL (always gateway mode)
sf = SmartflowClient("https://yourco.langsmart.app", api_key="sk-sf-...")

# 2. SMARTFLOW_GATEWAY_URL environment variable
# 3. ~/.smartflow/config.yaml  (written by `smartflow configure`)
# 4. Nothing configured -> direct mode, using OPENAI_API_KEY etc.
sf = SmartflowClient()

print(sf.mode)              # "gateway" or "direct"
print(sf.is_gateway_mode()) # True / False

First-run setup:

smartflow configure    # interactive wizard -> ~/.smartflow/config.yaml
smartflow status       # show mode, config source, connectivity
smartflow chat "Hello, which mode am I using?"

In direct mode the model prefix selects the provider (the same notation also works in gateway mode, where the proxy translates it):

await sf.chat("Hello", model="gpt-4o")                    # OpenAI
await sf.chat("Hello", model="anthropic/claude-sonnet-4-6")  # Anthropic
await sf.chat("Hello", model="gemini/gemini-2.0-flash")   # Gemini
await sf.chat("Hello", model="ollama/llama3")             # local Ollama
await sf.chat("Hello", model="local/my-fine-tuned-model") # LOCAL_BASE_URL

Gateway-only calls fail with a clear, actionable error:

from smartflow import SmartflowClient, DirectModeError

sf = SmartflowClient()          # direct mode
try:
    logs = await sf.get_logs()
except DirectModeError as e:
    print(e)  # names the feature + how to connect a gateway

Installation

pip install smartflow-sdk

Quick Start

Async Usage (Recommended)

import asyncio
from smartflow import SmartflowClient

async def main():
    async with SmartflowClient("http://your-smartflow:7775") as sf:
        # Simple chat
        response = await sf.chat("What is machine learning?")
        print(response)
        
        # Check cache stats
        stats = await sf.get_cache_stats()
        print(f"Cache hit rate: {stats.hit_rate:.1%}")
        print(f"Tokens saved: {stats.tokens_saved:,}")

asyncio.run(main())

Sync Usage

from smartflow import SyncSmartflowClient

sf = SyncSmartflowClient("http://your-smartflow:7775")

response = sf.chat("Explain quantum computing")
print(response)

sf.close()

OpenAI Drop-in Replacement

Just change the base_url - your existing OpenAI code works with Smartflow!

from openai import OpenAI

# Point to Smartflow instead of OpenAI
client = OpenAI(
    base_url="http://your-smartflow:7775/v1",
    api_key="your-key"  # Or use Smartflow's stored keys
)

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

Intelligent Compliance (ML-Powered)

Smartflow's adaptive learning compliance engine provides:

• Regex Pattern Matching - SSN, credit cards, emails, phone numbers, etc.
• ML Embedding Similarity - Semantic violation detection
• Behavioral Analysis - User pattern tracking and anomaly detection
• Organization Baselines - Deviation detection from org norms

async with SmartflowClient("http://your-smartflow:7775") as sf:
    # Scan content for compliance issues
    result = await sf.intelligent_scan(
        content="My SSN is 123-45-6789 and my email is john@example.com",
        user_id="user123",
        org_id="acme_corp"
    )
    
    print(f"Risk Score: {result.risk_score:.2f}")
    print(f"Risk Level: {result.risk_level}")
    print(f"Action: {result.recommended_action}")
    print(f"Explanation: {result.explanation}")
    
    # Check regex violations
    for violation in result.regex_violations:
        print(f"  - {violation['violation_type']}: {violation['severity']}")
    
    # Submit feedback to improve detection
    await sf.submit_compliance_feedback(
        scan_id="scan_abc123",
        is_false_positive=True,
        notes="This was a test number"
    )
    
    # Get learning status
    learning = await sf.get_learning_summary()
    print(f"Users tracked: {learning.total_users}")
    print(f"Learning complete: {learning.users_learning_complete}")
    
    # Get ML stats
    ml_stats = await sf.get_ml_stats()
    print(f"Total patterns: {ml_stats.total_patterns}")
    print(f"Learned patterns: {ml_stats.learned_patterns}")

Building AI Agents

Create AI agents with built-in compliance scanning and conversation memory:

from smartflow import SmartflowClient, SmartflowAgent

async with SmartflowClient("http://your-smartflow:7775") as sf:
    agent = SmartflowAgent(
        client=sf,
        name="CustomerSupport",
        model="gpt-4o",
        system_prompt="""You are a helpful customer support agent for TechCorp.
        Be professional, friendly, and always protect customer data.""",
        compliance_policy="enterprise_standard",
        enable_compliance_scan=True,
        user_id="support_agent_1",
        org_id="techcorp"
    )
    
    # Chat with automatic compliance scanning
    response = await agent.chat("How do I reset my password?")
    print(response)
    
    # Conversation memory is maintained
    response = await agent.chat("What about two-factor authentication?")
    print(response)
    
    # Get conversation history
    history = agent.get_history()
    print(f"Messages: {len(history)}")
    
    # Clear and start fresh
    agent.clear_history()

Workflow Orchestration

Chain AI operations with branching, parallel execution, and error handling:

from smartflow import SmartflowClient, SmartflowWorkflow

async with SmartflowClient("http://your-smartflow:7775") as sf:
    workflow = SmartflowWorkflow(sf, name="TicketClassification")
    
    # Step 1: Classify the ticket
    workflow.add_step(
        name="classify",
        action="chat",
        config={
            "prompt": "Classify this support ticket into one of: billing, technical, account. Ticket: {input}",
            "model": "gpt-4o-mini"
        },
        next_steps=["route"]
    )
    
    # Step 2: Route based on classification
    workflow.add_step(
        name="route",
        action="condition",
        config={
            "field": "output",
            "cases": {
                "billing": "billing_response",
                "technical": "technical_response",
                "account": "account_response"
            },
            "default": "general_response"
        }
    )
    
    # Execute the workflow
    result = await workflow.execute({"input": "My payment failed yesterday"})
    
    print(f"Success: {result.success}")
    print(f"Output: {result.output}")
    print(f"Steps executed: {result.steps_executed}")
    print(f"Execution time: {result.execution_time_ms:.1f}ms")

Monitoring & Analytics

async with SmartflowClient("http://your-smartflow:7775") as sf:
    # System health
    health = await sf.health_comprehensive()
    print(f"Status: {health.status}")
    print(f"Uptime: {health.uptime_seconds / 3600:.1f} hours")
    
    # Provider health
    providers = await sf.get_provider_health()
    for p in providers:
        print(f"{p.provider}: {p.status} ({p.latency_ms:.0f}ms)")
    
    # Cache statistics
    cache = await sf.get_cache_stats()
    print(f"Hit rate: {cache.hit_rate:.1%}")
    print(f"L1 hits: {cache.l1_hits}")
    print(f"L2 hits: {cache.l2_hits}")
    print(f"Tokens saved: {cache.tokens_saved:,}")
    
    # Audit logs
    logs = await sf.get_logs(limit=10)
    for log in logs:
        print(f"{log.timestamp}: {log.provider}/{log.model} - {log.tokens_used} tokens")

Configuration

Client Options

sf = SmartflowClient(
    base_url="http://smartflow:7775",      # Proxy URL (inference /v1/*)
    api_key="your-api-key",                 # Optional API key
    timeout=30.0,                           # Request timeout
    management_port=7778,                   # Management/admin API port
    compliance_port=7777,                   # Compliance API port
    bridge_port=3500,                       # Hybrid bridge port (historical logs)
    policy_perfect_port=7782,               # Policy Perfect builder port
)

Each Smartflow service listens on its own port; the SDK derives them all from the host in base_url:

Concern	Port	Example methods
Inference /v1/*	7775	chat, embeddings, image_generation, rag_query, call_mcp_tool
Management / admin	7778	list_mcp_servers, list_agents, issue_agent_credential, list_policies
Compliance scanning	7777	intelligent_scan, check_compliance
Policy Perfect builder	7782	assign_policy, generate_policies_from_document
Hybrid bridge	3500	get_logs_hybrid, get_analytics

Environment Variables

export SMARTFLOW_URL="http://your-smartflow:7775"
export SMARTFLOW_API_KEY="your-key"

API Reference

SmartflowClient Methods

Core AI & media

Method	Description
chat() / chat_completions()	Simple chat / OpenAI-compatible completions
stream_chat()	Streaming chat (async generator)
embeddings()	Generate text embeddings
image_generation()	Generate images (DALL·E / gpt-image-1)
audio_transcription() / text_to_speech()	Whisper ASR / TTS
rerank()	Cohere-compatible document reranking
claude_message()	Anthropic Claude Messages API
list_models()	List available models

Compliance, cache, monitoring

Method	Description
intelligent_scan()	ML-powered compliance scan
check_compliance() / redact_pii()	Rule-based scan / redaction
get_cache_stats()	Cache hit rates and savings
health() / readiness()	Proxy liveness / readiness probes
health_comprehensive()	Full system health
get_logs() / get_logs_hybrid()	VAS audit logs
get_provider_health()	Provider status

MCP tool servers

Method	Description
list_mcp_servers() / register_mcp_server() / remove_mcp_server()	MCP registry CRUD
discover_mcp_tools()	Probe a server URL for its tools (picker)
list_mcp_connectors() / list_mcp_skills()	Known connector presets / skills
get_mcp_catalog() / search_mcp_tools()	Browse / semantic-search the tool catalog
call_mcp_tool()	Governed JSON-RPC tools/call
get_mcp_trust() / evaluate_mcp_trust()	Trust registry
get_mcp_usage()	Per-server call counts and cost

Agents (A2A + AIDA)

Method	Description
list_agents() / register_agent() / remove_agent()	A2A registry CRUD
send_agent_task() / get_agent_task() / get_agent_card()	Invoke agents, poll tasks, read cards
issue_agent_credential() / verify_agent_credential() / revoke_agent_credential()	AIDA Ed25519 credentials (virtual ID)
get_aida_pubkey() / get_aida_jwks()	Offline credential verification keys

Policies & governance

Method	Description
list_policies() / create_policy() / resolve_policies()	Guardrail policy engine
list_policy_attachments() / attach_policy() / detach_policy()	Policy attachments
assign_policy() / list_policy_assignments()	Policy Perfect assignments by user/group/region
generate_policies_from_document()	Draft policies from a regulatory document
list_ai_inventory() / generate_examination_report()	AI inventory + examination/evidence reports
get_conformity_summary() / list_conformity_articles()	EU AI Act conformity
get_audit_logs() / verify_audit_chain()	Tamper-evident audit evidence
list_barriers() / get_barrier_attestation()	Information barriers

Retrieval & identity

Method	Description
create_vector_store() / search_vector_store()	Vector store CRUD + search
rag_ingest() / rag_query()	RAG ingest and query
get_sso_config() / set_sso_config() / list_sso_teams()	SSO / identity configuration

SmartflowAgent Methods

Method	Description
chat()	Chat with compliance scanning
clear_history()	Reset conversation
get_history()	Get conversation history

SmartflowWorkflow Methods

Method	Description
add_step()	Add a workflow step
set_entry()	Set entry point
execute()	Run the workflow

Examples

See the examples/ directory for more:

• simple_chat.py - Basic chat usage
• compliance_check.py - PII detection and redaction
• system_monitoring.py - Health and analytics
• openai_drop_in.py - OpenAI compatibility
• agent_example.py - Building AI agents
• workflow_example.py - Workflow orchestration

License

MIT License - see LICENSE for details.

Support

• SDK reference: https://docs.aperion.ai/smartflow-sdk-reference.html
• API reference: https://docs.aperion.ai/smartflow-api-reference.html
• Email: support@langsmart.app

---

Built with ❤️ by Langsmart, Inc.