RelationalAI Agent SDK

Python client SDK for interacting with RelationalAI's Agent API. Build intelligent agents that can query, analyze, and reason over your semantic data models.

Overview

The RelationalAI Agent SDK provides a high-level Python interface for:

• Natural Language Queries: Ask questions in plain English about your data models
• Semantic Model Integration: Work with PyRel concepts, relationships, and data sources
• Streaming Responses: Receive real-time progress updates and results
• Capability Control: Fine-grained control over agent capabilities

Installation

pip install relationalai-agent

Quick Start

Basic Usage

from relationalai_agent import Agent, AgentConfig
from relationalai.semantics import Model, Concept, Relationship

# Create your PyRel model
model = Model("my_model")
Customer = model.Concept("Customer")
Customer.name = Relationship("{Customer} has {name:String}")
Customer.email = Relationship("{Customer} has {email:String}")

# Configure and create agent
config = AgentConfig(endpoint="https://agent-api.relational.ai/ask")
agent = Agent(config)

# Register your model
agent.use(Customer, Customer.name, Customer.email)

# Ask questions
response = await agent.ask("What are all the customer names?")

# Get results
for result in response.results:
    if result.type == "prose":
        answer = await result.fetch()
        print(answer)

Streaming Responses

# Stream events as they arrive
async for event in agent.ask("Show top 10 customers by revenue").stream():
    if event.type == "status":
        print(f"Status: {event.message}")
    elif event.type == "plan":
        print(f"Plan: {len(event.tasks)} tasks")
    elif event.type == "execute_code":
        print(f"Generated code:\n{event.code}")
    elif event.type == "result":
        result = await event.result.fetch()
        print(f"Result: {result}")

Features

Agent Configuration

from relationalai_agent import Agent, AgentConfig, CapabilityConfig, Quotas

# Configure capabilities
capabilities = CapabilityConfig(
    query=True,      # Allow querying data
    optimize=False,  # Disable optimization
    predict=False,   # Disable predictions
    visualize=True   # Allow visualizations
)

# Set resource quotas
quotas = Quotas(
    max_execution_time=60,    # seconds
    max_result_size=1000000,  # bytes
    max_iterations=10         # max planning iterations
)

# Create configured agent
config = AgentConfig(
    endpoint="https://agent-api.relational.ai/ask",
    capabilities=capabilities,
    quotas=quotas
)
agent = Agent(config)

Context Management

# Add context
with agent.use(Customer, Order, customer_orders) as context:
    # Context is active here
    response = await agent.ask("What's the average order value?")

# Context is automatically removed after the block

Event Types

The SDK provides typed event objects for all streaming events:

• StatusEvent: Progress updates (event.message)
• PlanEvent: Query plan details (event.tasks)
• TaskStartEvent: Task begin notification (event.task_id, event.description)
• TaskEndEvent: Task completion (event.task_id, event.status)
• ExecuteCodeEvent: Generated code (event.code, event.language)
• ResultEvent: Final results (event.result)
• ErrorEvent: Error details (event.error_message, event.details)

Result Types

Results come in different forms:

# Data result (DataFrame)
if result.type == "data":
    df = await result.fetch()
    print(df.head())

# Prose result (text explanation)
if result.type == "prose":
    text = await result.fetch()
    print(text)

PyRel Integration

The SDK seamlessly integrates with PyRel semantic models:

from relationalai.semantics import Model, Concept, Relationship

# Define your model
model = Model("ecommerce")

# Define concepts
Customer = model.Concept("Customer")
Product = model.Concept("Product")
Order = model.Concept("Order")

# Define relationships
Customer.name = Relationship("{Customer} has {name:String}")
Product.price = Relationship("{Product} has {price:Float}")
customer_orders = Relationship("{customer:Customer} placed {order:Order}")

# Register with agent
agent.use(Customer, Product, Order, customer_orders)

# Ask natural language questions
response = await agent.ask("Which customers ordered products over $100?")

Advanced Usage

Custom Timeout and Retry

import httpx

# Create custom HTTP client with timeout
timeout = httpx.Timeout(connect=10.0, read=300.0, write=10.0, pool=10.0)
# Note: Transport configuration is internal, configure at AgentConfig level

Error Handling

async for event in agent.ask("Complex query").stream():
    if event.type == "error":
        print(f"Error: {event.error_message}")
        if event.details:
            print(f"Details: {event.details}")
        # Handle error appropriately

Multiple Contexts

# Add different contexts for different queries
agent.use(Customer, customer_context_items)

response1 = await agent.ask("Customer analytics query")

agent.use(Product, product_context_items)

response2 = await agent.ask("Product analytics query")

# Or combine contexts
agent.use(Customer, Product, Order, customer_orders)
response3 = await agent.ask("Combined customer-product query")

Requirements

• Python 3.11+
• relationalai >= 0.12
• relationalai-agent-shared >= 0.1.0
• pydantic >= 2.0.0
• httpx >= 0.27.0

API Reference

Agent

• Agent(config: AgentConfig) - Create new agent instance
• agent.use(*context) - Register context items (PyRel objects or ECS components)
• agent.ask(question, **kwargs) - Ask a question, returns awaitable Question
• agent.drop(*context) - Remove context items

AgentConfig

• endpoint: str - Agent API endpoint URL
• capabilities: CapabilityConfig | None - Capability restrictions
• quotas: Quotas | None - Resource quotas

Question

• await question - Get Response with all results
• question.stream() - Get StreamResponse for real-time events

Response

• results: list[Result] - All results from the query
• usage: Usage - Resource usage information

Examples

See the examples/ directory for complete working examples.

License

Apache License 2.0 - see LICENSE file for details.

Support

• Documentation: https://docs.relational.ai
• Issues: https://github.com/RelationalAI/relationalai-agent/issues
• Email: support@relational.ai

Related Packages

• relationalai - RelationalAI Python SDK
• relationalai-agent-shared - Shared types and contracts

---

Note: This package requires access to RelationalAI's Agent API service. Contact sales@relational.ai for access.