Python SDK

pip install engrami

poetry add engrami

from engrami import EngramiClient

# Initialize with API key
client = EngramiClient(api_key="your-api-key")

# Or use environment variable ENGRAMI_API_KEY
client = EngramiClient()

# Create a customer support agent
agent = client.agents.create(
    name="Support Agent",
    type="support",
    description="Handles customer inquiries",
    memory_types=["semantic", "episodic"],
    config={
        "model": "gpt-4-turbo",
        "temperature": 0.7,
        "personality": "helpful and professional"
    }
)

print(f"Created agent: {agent.id}")

# Send a message and get a response
response = agent.chat(
    message="How do I reset my password?",
    session_id="user-123"  # Optional: for conversation continuity
)

print(response.content)
print(f"Tokens used: {response.usage.total_tokens}")

# List all agents
agents = client.agents.list()

# Get a specific agent
agent = client.agents.get("agent-id")

# Update an agent
agent = client.agents.update(
    agent_id="agent-id",
    name="Updated Name",
    config={"temperature": 0.8}
)

# Delete an agent
client.agents.delete("agent-id")

# Add knowledge to semantic memory
client.memory.add(
    agent_id="agent-id",
    memory_type="semantic",
    content="Our refund policy allows returns within 30 days.",
    metadata={"category": "policy", "source": "faq"}
)

# Search semantic memory
results = client.memory.search(
    agent_id="agent-id",
    memory_type="semantic",
    query="refund policy",
    top_k=5
)

for result in results:
    print(f"Score: {result.score}, Content: {result.content}")

# Create a workflow
workflow = client.workflows.create(
    name="Support Escalation",
    description="Handle customer support with escalation",
    nodes=[
        {
            "id": "start",
            "type": "trigger",
            "config": {"trigger_type": "webhook"}
        },
        {
            "id": "classify",
            "type": "agent",
            "config": {"agent_id": "classifier-agent-id"}
        },
        {
            "id": "respond",
            "type": "agent",
            "config": {"agent_id": "support-agent-id"}
        }
    ],
    edges=[
        {"source": "start", "target": "classify"},
        {"source": "classify", "target": "respond"}
    ]
)

# Execute a workflow
result = client.workflows.execute(
    workflow_id=workflow.id,
    input={"message": "I need help with my order"}
)

import asyncio
from engrami import AsyncEngramiClient

async def main():
    client = AsyncEngramiClient(api_key="your-api-key")

    # Create agent asynchronously
    agent = await client.agents.create(
        name="Async Agent",
        type="assistant"
    )

    # Chat asynchronously
    response = await agent.chat("Hello!")
    print(response.content)

    # Streaming responses
    async for chunk in agent.chat_stream("Tell me a story"):
        print(chunk.content, end="", flush=True)

asyncio.run(main())

from engrami.exceptions import (
    EngramiError,
    AuthenticationError,
    RateLimitError,
    NotFoundError,
    ValidationError
)

try:
    agent = client.agents.get("non-existent-id")
except NotFoundError:
    print("Agent not found")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
except EngramiError as e:
    print(f"API error: {e.message}")

from engrami import EngramiClient

client = EngramiClient(
    api_key="your-api-key",
    base_url="https://api.engrami.com",  # Custom API endpoint
    timeout=30.0,  # Request timeout in seconds
    max_retries=3,  # Retry failed requests
    debug=True  # Enable debug logging
)