Teleon

# Install
pip install teleon

# Create an agent
from teleon import TeleonClient

client = TeleonClient(api_key="tlk_live_...")

@client.agent(
    name="customer-support",
    cortex=True,
    helix={"min_instances": 1, "max_instances": 100},
    sentinel={"content_filtering": True, "pii_detection": True}
)
async def support_agent(customer_message: str, customer_id: str, cortex) -> str:
    """Handle customer support with AI."""
    return await process_support_request(customer_message)

# Deploy
$ teleon deploy

# Done! 🚀

from teleon import TeleonClient

client = TeleonClient(api_key="tlk_live_...")

@client.agent(name="hello-world")
async def hello_agent(name: str) -> str:
    """A simple greeting agent."""
    return f"Hello, {name}! Welcome to Teleon. 🚀"

@client.agent(
    name="support",
    cortex={
        "auto": True,
        "scope": ["customer_id"],        # Automatic multi-tenant isolation
        "auto_context": {
            "enabled": True,
            "history_limit": 10,         # Last N conversations
            "relevant_limit": 5,         # Top N semantic matches
            "max_tokens": 2000
        }
    }
)
async def support_agent(query: str, customer_id: str, cortex):
    # Search for relevant past interactions
    results = await cortex.search(
        query="previous issues",
        filter={"type": "resolution"}
    )

    # Get recent history
    history = await cortex.get(filter={}, limit=10)

    # Store new information
    await cortex.store(
        content=f"Resolved: {query}",
        type="resolution"
    )

    return await generate_response(query, results, history)

@client.agent(
    name="assistant",
    cortex={
        "scope": ["org_id", "user_id"],
        "layers": {
            "company": {"scope": []},           # Shared across company
            "team": {"scope": ["org_id"]},      # Scoped to team
            "personal": {"scope": ["user_id"]}  # Scoped to user
        }
    }
)
async def assistant(query: str, org_id: str, user_id: str, cortex):
    company_docs = await cortex.company.search(query=query, limit=5)
    personal_notes = await cortex.personal.get(filter={}, limit=3)
    ...

from teleon.cortex import set_storage_backend, PostgresBackend

backend = PostgresBackend(host="localhost", database="teleon", user="postgres", password="secret")
set_storage_backend(backend)

@client.agent(
    name="production-agent",
    helix={
        "min_instances": 2,
        "max_instances": 50,
        "cpu_limit": 2.0,
        "memory_limit_mb": 512,
        "health_check_enabled": True,
        "health_check_interval": 30,
        "startup_timeout": 30,
        "shutdown_timeout": 30
    }
)
async def production_agent(query: str) -> dict:
    return await process(query)

@client.agent(
    name="my-agent",
    helix={
        "min": 2,       # min_instances
        "max": 10,      # max_instances
        "cpu": 2.0,     # cpu_limit
        "memory": 512   # memory_limit_mb
    }
)

from teleon.helix import AgentRuntime, RuntimeConfig, ResourceConfig

runtime = AgentRuntime(RuntimeConfig(environment="production", max_workers=20))

await runtime.register_llm_agent(
    agent_id="chat-agent",
    agent_callable=chat_handler,
    model="gpt-4",
    max_tokens=2000,
    cost_budget=10.0,          # $10/hour budget
    resources=ResourceConfig(min_instances=3, max_instances=20)
)

from teleon.helix import create_cache, CacheStrategy, CacheEvictionPolicy

cache = create_cache(
    max_size=1000,
    strategy=CacheStrategy.EXACT,
    eviction_policy=CacheEvictionPolicy.LRU,
    default_ttl=3600
)

cached = await cache.get(messages, model="gpt-4")
if not cached:
    response = await llm.complete(messages, model="gpt-4")
    await cache.set(messages=messages, model="gpt-4", response=response, ttl=3600)

stats = cache.get_statistics()
print(f"Hit rate: {stats['hit_rate']*100:.1f}%")
print(f"Tokens saved: {stats['total_tokens_saved']:,}")

from fastapi import FastAPI
from teleon.helix import setup_health_endpoints

app = FastAPI()
health_manager = setup_health_endpoints(app, service_name="my-agent", version="1.0.0")

async def check_database():
    await db.ping()
    return True, "Database connected"

health_manager.add_check("database", check_database, critical=True)
health_manager.set_ready(True, reason="Initialization complete")

@client.agent(
    name="customer-support",
    sentinel={
        "enabled": True,
        "content_filtering": True,        # Toxicity, hate speech, profanity
        "pii_detection": True,            # Emails, phones, SSNs, credit cards, IPs
        "compliance": ["gdpr", "hipaa"],  # Compliance standards
        "moderation_threshold": 0.8,      # Sensitivity (0.0–1.0)
        "action_on_violation": "block",   # block | flag | redact | escalate
        "log_violations": True,
        "audit_enabled": True
    }
)
async def support_agent(query: str) -> str:
    # Sentinel validates input before execution and output after
    return await handle_query(query)

from teleon.sentinel.policy_engine import PolicyEngine

policy_engine = PolicyEngine()

policy_engine.add_policy("no_competitors", {
    "type": "regex",
    "pattern": r"\b(CompetitorA|CompetitorB)\b",
    "message": "Competitor mention not allowed",
    "severity": "high"
})

policy_engine.add_policy("min_response_length", {
    "type": "condition",
    "condition": "len(text) < 50",
    "message": "Response too short",
    "severity": "low"
})

audit_logger = engine.get_audit_logger()

stats = audit_logger.get_violation_stats(agent_id="my-agent")
# {
#   "total_violations": 42,
#   "by_type": {"pii_detection": 30, "toxicity": 12},
#   "by_action": {"block": 35, "flag": 7}
# }

json_export = audit_logger.export_audit_trail(format="json")
csv_export = audit_logger.export_audit_trail(format="csv")

from teleon.llm import LLMGateway
from teleon.llm.providers import OpenAIProvider, AnthropicProvider

gateway = LLMGateway()
gateway.register_provider(OpenAIProvider(api_key="..."))
gateway.register_provider(AnthropicProvider(api_key="..."))

# Simple chat
response = await gateway.chat(message="What is AI?", model="gpt-4")

# Function calling
response = await gateway.chat_with_functions(
    message="Get weather in Paris",
    functions=[get_weather_function],
    model="gpt-4"
)

# Streaming
async for chunk in gateway.stream(message="Write a story", model="gpt-4"):
    print(chunk, end="")

@client.agent(
    name="tooled-agent",
    tools=["http_request", "web_scraper", "send_email", "sql_query"]
)
async def tooled_agent(task: str) -> dict:
    return await execute_with_tools(task)

@client.agent(
    name="monitored-agent",
    observability={
        "tracing": True,
        "metrics": True,
        "logging": "detailed"
    }
)
async def monitored_agent(input: str) -> str:
    return await process(input)

from teleon.helix import init_agent_reporter

reporter = await init_agent_reporter(
    deployment_id="deploy-123",
    api_key="tlk_xxx",
    flush_interval=10.0
)

await reporter.report_request(
    input_tokens=100,
    output_tokens=50,
    latency_ms=250.5,
    model="gpt-4",
    success=True,
    cost=0.002
)

@client.agent(
    name="support-bot",
    cortex={
        "auto": True,
        "scope": ["customer_id"],
        "auto_context": {"enabled": True, "history_limit": 20, "relevant_limit": 5}
    },
    helix={"min_instances": 2, "max_instances": 10},
    sentinel={"pii_detection": True, "action_on_violation": "redact"},
    tools=["sql_query", "send_email", "slack_message"]
)
async def support_bot(query: str, customer_id: str, cortex) -> dict:
    # Auto-context injects relevant history before execution
    context = cortex.context.text if cortex.context else ""

    similar = await cortex.search(query=query, filter={"type": "resolution"}, limit=3)

    response = await llm.chat(
        f"Customer query: {query}\nContext: {context}\nSimilar resolutions: {similar}"
    )

    if "resolved" in response.lower():
        await cortex.store(
            content=f"Resolved: {query} -> {response}",
            type="resolution"
        )

    return {"status": "resolved", "response": response}

@client.agent(
    name="healthcare-assistant",
    sentinel={
        "pii_detection": True,
        "compliance": ["hipaa", "gdpr"],
        "action_on_violation": "block",
        "audit_enabled": True
    }
)
async def healthcare_agent(query: str, patient_id: str):
    # All PHI is blocked; every violation is logged for HIPAA audit
    return await generate_response(query)

@client.agent(
    name="payments",
    sentinel={
        "pii_detection": True,
        "compliance": ["pci_dss"],
        "action_on_violation": "block"
    }
)
async def payment_agent(request: dict):
    return await process_payment(request)

@client.agent(name="data-collector", tools=["web_scraper", "http_request"])
async def data_collector(topic: str) -> dict:
    results = []
    for source in get_sources(topic):
        results.append(await tools.http_request(url=source))
    return {"data": results}

@client.agent(name="analyzer", model="gpt-4")
async def analyzer(data: dict) -> dict:
    analysis = await llm.chat(f"Analyze this data: {data}", model="gpt-4")
    return {"analysis": analysis}

@client.agent(name="research-orchestrator")
async def orchestrator(research_topic: str) -> dict:
    data_result = await data_collector(research_topic)
    analysis_result = await analyzer(data_result)
    return {"topic": research_topic, "analysis": analysis_result["analysis"]}

@client.agent(
    name="chat",
    cortex={"scope": ["session_id"], "auto": False}
)
async def chat_agent(message: str, session_id: str, cortex):
    context = await cortex.get(filter={}, limit=10)
    response = await llm.chat(message, context)

    # Expire session memory after 1 hour
    await cortex.store(
        content=f"User: {message}\nAssistant: {response}",
        type="message",
        ttl=3600
    )
    return response

from teleon.scheduler import schedule
from teleon.webhooks import webhook

@client.agent(name="daily-report")
@schedule(cron="0 9 * * *")
async def daily_report() -> dict:
    metrics = await gather_daily_metrics()
    report = await generate_report(metrics)
    await send_email_report(report)
    return {"status": "sent"}

@client.agent(name="github-bot")
@webhook(path="/github", methods=["POST"])
async def github_bot(event: dict) -> dict:
    if event["type"] == "pull_request":
        await review_pr(event["pr_number"])
    elif event["type"] == "issue":
        await triage_issue(event["issue_number"])
    return {"status": "processed"}

from teleon import TeleonClient

client = TeleonClient(
    api_key="tlk_live_...",
    environment="production",  # "production" | "staging" | "dev"
    base_url=None,
    verify_key=True
)

@client.agent(
    name: str,                    # Agent name (required)
    description: str = None,
    model: str = "gpt-3.5-turbo",
    temperature: float = 0.7,
    tools: list[str] = None,
    helix: bool | dict = None,    # Helix runtime config
    cortex: bool | dict = False,  # Cortex memory config
    sentinel: bool | dict = None, # Sentinel safety config
    observability: dict = None,
)

# store
entry_id = await cortex.store(content="...", ttl=3600, upsert=True, **custom_fields)

# search (semantic)
results = await cortex.search(query="...", filter={"field": "value"}, limit=10)

# get (filter only, no semantic)
entries = await cortex.get(filter={"field": "value"}, limit=50)

# update
count = await cortex.update(filter={"field": "value"}, content="new", **new_fields)

# delete
deleted = await cortex.delete(filter={"user_id": "alice"})

# count
n = await cortex.count(filter={"type": "query"})

# auto-injected context
cortex.context.entries   # List of Entry objects
cortex.context.text      # Formatted text for LLM injection

entry.id          # Unique identifier
entry.content     # Text content
entry.fields      # Dict of all custom fields
entry.created_at  # Creation timestamp
entry.updated_at  # Last update timestamp
entry.expires_at  # Expiration (if TTL set)
entry.score       # Relevance score (search only)

from teleon.helix import AgentRuntime, RuntimeConfig, ResourceConfig

runtime = AgentRuntime(RuntimeConfig(environment="production", max_workers=20))

# Standard agent
await runtime.register_agent(agent_id, agent_callable, resources=ResourceConfig(...))

# LLM agent with cost tracking
await runtime.register_llm_agent(agent_id, agent_callable, model="gpt-4", cost_budget=10.0)

await runtime.start_agent("my-agent")
await runtime.stop_agent("my-agent", force=False)
await runtime.scale_agent("my-agent", instances=5)

status = await runtime.get_agent_status("my-agent")
# {"status": "running", "instances": 3, "health": "healthy", ...}

from teleon.sentinel import SentinelEngine, SentinelConfig, ComplianceStandard, GuardrailAction

config = SentinelConfig(
    enabled=True,
    content_filtering=True,
    pii_detection=True,
    compliance=[ComplianceStandard.GDPR, ComplianceStandard.HIPAA],
    action_on_violation=GuardrailAction.BLOCK
)

engine = SentinelEngine(config)

result = await engine.validate_input(user_input, agent_name="my-agent")
result = await engine.validate_output(agent_output, agent_name="my-agent")

result.passed             # bool
result.action             # GuardrailAction
result.violations         # List[Dict]
result.redacted_content   # Optional[str]

from teleon.integrations.azure import AzureOpenAIProvider, AzureStorageClient
from teleon.integrations.aws import BedrockProvider, S3Client, DynamoDBClient
from teleon.integrations.gcp import VertexAIProvider, CloudStorageClient

from teleon.integrations.slack import SlackClient
from teleon.integrations.discord import DiscordClient
from teleon.integrations.email import EmailClient

from teleon.integrations.database import PostgresClient, MongoClient
from teleon.integrations.cache import RedisClient

from teleon.testing import AgentTestCase, MockLLM, MockTool

class TestMyAgent(AgentTestCase):
    async def test_agent_response(self):
        with MockLLM(response="Mocked response"):
            result = await my_agent("test input")
            assert result == "Mocked response"

    async def test_agent_with_tools(self):
        with MockTool("http_request", return_value={"status": 200}):
            result = await my_agent("fetch data")
            assert "status" in result

from teleon.testing import load_test

results = await load_test(
    agent=my_agent,
    input_generator=lambda: {"query": "test"},
    concurrent_users=100,
    duration_seconds=60
)

print(f"RPS: {results.requests_per_second}")
print(f"P99 latency: {results.p99_latency_ms}ms")
print(f"Error rate: {results.error_rate}%")