Teleon Library

Overview

Teleon is a production-ready platform for building, deploying, and managing AI agents.

Role

The Teleon Python library is the primary integration surface for your application code. It provides:

Authentication and agent registration via TeleonClient
The @client.agent(...) decorator for defining agents (async functions)
The @tool(...) decorator for defining reusable tools
Optional integration points for Cortex (memory), Sentinel (safety), and Helix (runtime)

When to use

You want the exact API surface (classes, decorators, methods) and configuration defaults.
You need patterns like scope checks (has_scope / require_scope) or working with remote/deployed agents.

Use this API reference when you need:

The exact Python API surface (classes, decorators, methods)
Configuration options and defaults
Exception types and error-handling patterns

Installation

pip install teleon

Imports

from teleon import TeleonClient

Authentication

API key formats

API keys must follow one of these formats:

tlk_live_xxx (production)
tlk_test_xxx (test)
teleon_xxx (legacy)

Environment variables

Variable	Description
`TELEON_API_KEY`	Your Teleon API key
`TELEON_API_URL`	Custom API URL
`TELEON_PLATFORM_URL`	Platform URL for metrics
`TELEON_QUIET`	Suppress initialization messages
`TELEON_METRICS_ENABLED`	Enable/disable metrics reporting

`TeleonClient`

Core class for authenticating and registering agents.

Initialization

from teleon import TeleonClient
 
client = TeleonClient(
    api_key="tlk_live_xxx",        # Required
    environment="production",      # Optional: production, staging, dev (default: production)
    base_url=None,                  # Optional: Custom API base URL
    verify_key=True,                # Optional: Verify API key with backend (default: True)
    is_paid_tier=False              # Optional: Paid tier flag (affects embedding model)
)

Methods

Method	Description
`agent()`	Decorator to register an agent
`has_scope(*scopes)`	Check if API key has required scopes
`require_scope(*scopes)`	Raise error if key lacks scopes
`get_remote_agent(name)`	Get reference to a deployed agent
`list_deployed_agents()`	List all deployed agents
`close()`	Close HTTP client connections

Class methods

# Get all registered agents across all clients
agents = TeleonClient.get_all_agents()
 
# Get a specific agent by ID
agent = TeleonClient.get_agent(agent_id)
 
# Get a client by user ID
client = TeleonClient.get_client(user_id)

Agent decorator: `@client.agent(...)`

Registers an agent with the Teleon platform.

Requirements

Agent functions must be async def
Agent functions must return a value

Basic usage

@client.agent(name="support-agent")
async def support_agent(query: str) -> str:
    return "Response"

Full configuration

@client.agent(
    name="support",                    # Required: Agent name
    description="Customer support",    # Optional: Agent description
    model="gpt-4",                     # Optional (default: gpt-4)
    temperature=0.7,                   # Optional (default: 0.7)
    max_tokens=500,                    # Optional (default: 500)
    helix={                            # Optional: Helix runtime configuration
        "min": 1,
        "max": 10,
        "target_cpu": 70
    },
    cortex=True,                       # Optional: Enable Cortex memory
    sentinel={                         # Optional: Sentinel safety configuration
        "enabled": True,
        "pii_detection": True
    }
)
async def support_agent(query: str, customer_id: str, cortex) -> str:
    return "response"

Configuration options

Option	Type	Default	Description
`name`	str	required	Agent name
`description`	str	function docstring	Agent description
`model`	str	`"gpt-4"`	LLM model to use
`temperature`	float	`0.7`	Temperature setting
`max_tokens`	int	`500`	Maximum tokens
`helix`	dict	`None`	Helix runtime config
`cortex`	bool/dict	`None`	Cortex memory config
`sentinel`	dict	`None`	Sentinel safety config
`tools`	list	`[]`	Tools available to agent
`timeout`	int	`None`	Execution timeout

Tool decorator: `@tool(...)`

Defines a reusable tool that agents can execute.

Basic usage

from teleon import tool
 
@tool(name="calculator", category="math")
async def calculate(expression: str) -> float:
    return eval(expression)

Full configuration

@tool(
    name="web-search",            # Optional: Tool name (default: function name)
    description="Search the web", # Optional: Description (default: docstring)
    category="search",            # Optional: Tool category
    cache_ttl=3600,                # Optional: Cache results (seconds)
    timeout=30.0                   # Optional: Execution timeout
)
async def web_search(query: str, num_results: int = 5) -> dict:
    return {"results": []}

Configuration options

Option	Type	Default	Description
`name`	str	function name	Tool name
`description`	str	docstring	Tool description
`category`	str	`"general"`	Tool category
`cache_ttl`	int	`None`	Cache TTL in seconds
`timeout`	float	`None`	Execution timeout

Tool registry

from teleon.tools import get_registry
from teleon.tools import ToolRegistry
 
registry = get_registry()
 
# Register a tool
await registry.register(tool)
 
# Get a tool by name
tool = registry.get("calculator")
 
# List all tools
tools = registry.list_tools(category=ToolCategory.MATH)
 
# Search tools
results = registry.search("calculate")
 
# Get registry stats
stats = registry.get_stats()

Remote agents

Interact with deployed agents from your applications.

Getting a remote agent

client = TeleonClient(api_key="tlk_live_xxx")
 
agent = await client.get_remote_agent("customer-support")
agents = await client.list_deployed_agents()

Executing a remote agent

result = await agent.execute(
    input_data={"query": "Help me with billing"},
    temperature=0.7,
    max_tokens=500,
    metadata={"session_id": "123"}
)
 
print(result["response"])
print(result["metrics"])
 
results = await agent.execute_batch(
    inputs=[
        {"query": "Question 1"},
        {"query": "Question 2"},
        {"query": "Question 3"}
    ],
    parallel=True
)

Remote agent management

info = await agent.get_info()
metrics = await agent.get_metrics(period="24h")
logs = await agent.get_logs(limit=100, level="ERROR")
 
await agent.update_config({
    "temperature": 0.5,
    "max_tokens": 1000
})
 
await agent.scale(replicas=5)

Remote Cortex access

results = await agent.cortex.search(
    query="billing issues",
    limit=10
)
 
await agent.cortex.store(
    content="Important information",
    metadata={"type": "note"}
)
 
stats = await agent.cortex.get_stats()

LLM Gateway

Unified interface for multiple LLM providers.

Quick start

from teleon.llm.gateway import configure_gateway
 
gateway = configure_gateway(
    openai_api_key="sk-xxx",
    anthropic_api_key="sk-ant-xxx",
    enable_cache=True,
    cache_ttl=3600
)
 
from teleon.llm import LLMMessage, LLMConfig
 
messages = [LLMMessage(role="user", content="Hello!")]
 
config = LLMConfig(
    model="gpt-4",
    temperature=0.7,
    max_tokens=500,
    use_cache=True
)
 
response = await gateway.complete(messages, config)
print(response.content)

Streaming

async for chunk in gateway.stream_complete(messages, config):
    print(chunk, end="")

Supported providers

Provider	Models
OpenAI	gpt-4, gpt-4-turbo, gpt-4o, gpt-3.5-turbo
Anthropic	claude-3-opus, claude-3-sonnet, claude-3-haiku
Google	gemini-pro, gemini-1.5-pro
Azure OpenAI	All Azure-deployed models

Statistics

stats = gateway.get_stats()
print(f"Total cost: ${stats['total_cost']:.4f}")
print(f"Total tokens: {stats['total_tokens']}")
print(f"Success rate: {stats['success_rate']:.2%}")

Integrations

Pre-built integrations for popular services.

Slack

from teleon.integrations.slack import SlackIntegration
from teleon.integrations.base import IntegrationConfig
 
config = IntegrationConfig(
    name="slack",
    api_key="xoxb-your-bot-token"
)
 
slack = SlackIntegration(config)
await slack.authenticate()
 
await slack.send_message("#general", "Hello from Teleon!")
 
await slack.send_message(
    "#general",
    text="Hello!",
    blocks=[{"type": "section", "text": {"type": "mrkdwn", "text": "*Bold text*"}}]
)
 
channel = await slack.create_channel("new-channel", is_private=False)
 
await slack.upload_file(
    channels=["#general"],
    content="File content",
    filename="report.txt"
)
 
user = await slack.get_user_info("U1234567")
 
await slack.close()

GitHub

from teleon.integrations.github import GitHubIntegration
from teleon.integrations.base import IntegrationConfig
 
config = IntegrationConfig(
    name="github",
    api_key="ghp_your_token"
)
 
github = GitHubIntegration(config)
await github.authenticate()
 
issue = await github.create_issue(
    repo="owner/repo",
    title="Bug found",
    body="Description of the bug",
    labels=["bug"],
    assignees=["username"]
)
 
pr = await github.create_pull_request(
    repo="owner/repo",
    title="Feature: New feature",
    head="feature-branch",
    base="main",
    body="PR description"
)
 
repos = await github.list_repositories(org="my-org")
commit = await github.get_commit("owner/repo", "abc123")
 
webhook = await github.create_webhook(
    repo="owner/repo",
    url="https://example.com/webhook",
    events=["push", "pull_request"]
)
 
await github.close()

Testing framework

Base test classes

from teleon.testing import AgentTestCase, ToolTestCase, IntegrationTestCase
 
class MyAgentTest(AgentTestCase):
    async def test_agent_response(self):
        result = await self.run_agent("Hello")
        self.assert_agent_output(result, contains="response")
 
class MyToolTest(ToolTestCase):
    async def test_tool_execution(self):
        result = await self.run_tool("calculator", "2+2")
        assert result == 4

Mocks

from teleon.testing import MockLLM, MockTool, MockMemory, MockMessageBus
 
mock_llm = MockLLM()
mock_llm.set_response("Expected response")
 
mock_tool = MockTool()
mock_tool.set_result({"data": "value"})
 
mock_memory = MockMemory()
mock_memory.add_entry("test content", customer_id="123")
 
mock_bus = MockMessageBus()

Assertions

from teleon.testing import (
    assert_agent_output,
    assert_tool_called,
    assert_memory_stored,
    assert_cost_below
)
 
assert_agent_output(result, contains="expected")
assert_agent_output(result, matches=r"pattern.*")
 
assert_tool_called("calculator", with_args={"expression": "2+2"})
assert_memory_stored(content="test", customer_id="123")
assert_cost_below(result, max_cost=0.01)

Errors

Teleon provides an exception hierarchy to support robust error handling.

Exception types

from teleon.core import (
    TeleonError,
 
    ConfigurationError,
 
    AgentError,
    AgentNotFoundError,
    AgentExecutionError,
    AgentValidationError,
 
    LLMError,
    LLMProviderError,
    LLMTimeoutError,
    LLMRateLimitError,
    LLMContextLengthError,
 
    ToolError,
    ToolNotFoundError,
    ToolExecutionError,
    ToolValidationError,
    ToolTimeoutError,
    ToolPermissionError,
 
    MemoryError,
 
    SecurityError,
    AuthenticationError,
    AuthorizationError,
    InvalidAPIKeyError,
 
    ResourceError,
    QuotaExceededError,
    RateLimitExceededError,
)

Error-handling pattern

from teleon.core import TeleonError, LLMRateLimitError, AgentExecutionError
 
try:
    result = await my_agent("query")
except LLMRateLimitError as e:
    await asyncio.sleep(e.retry_after)
    result = await my_agent("query")
except AgentExecutionError as e:
    logging.error(f"Agent failed: {e}")
except TeleonError as e:
    logging.error(f"Teleon error: {e}")

Teleon Library

On this page