agentstable

AgentStable SDK

A Python SDK for integrating with the AgentStable Search Action Service. This SDK helps AI agents find and execute appropriate API actions based on natural language queries using the agents.json schema.

Installation

pip install agentstable

Quick Start

Using with OpenAI

import agentstable
from openai import OpenAI

# 1. Search for actions that match a natural language query
query = "Create a product called Premium Access for $100"
flow = agentstable.search(
    query=query,
    collection_id="your_collection_id",  # Optional
    base_url="http://localhost:8081/api"  # Change to your service URL
)

# 2. Generate OpenAI tools from the flow
tools = agentstable.get_tools(flow)

# 3. Use the tools with an LLM
client = OpenAI()
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{
        "role": "user",
        "content": query
    }],
    tools=tools
)

# 4. Execute the API calls using the LLM's output
auth = agentstable.BearerAuth(token="your_api_token")
result = agentstable.execute(flow, response, auth)

print(f"Execution results: {result}")

Using with Anthropic

import agentstable
from anthropic import Anthropic

# 1. Search and generate Anthropic tools in one step
query = "Plan a vacation to Hawaii"
result = agentstable.search_and_generate_anthropic_tools(
    query=query,
    collection_id="your_collection_id",  # Optional
)

flow = result["flow"]
tools = result["tools"]

# 2. Use the tools with Anthropic
client = Anthropic()
response = client.messages.create(
    model="claude-3-sonnet-20240229",
    messages=[{
        "role": "user",
        "content": query
    }],
    tools=tools
)

# 3. Execute the API calls using the Claude's output
auth = agentstable.BearerAuth(token="your_api_token")
result = agentstable.execute_anthropic(flow, response, auth)

print(f"Execution results: {result}")

Comprehensive Example with All Features

This example demonstrates the full range of SDK features, including:

• Memory/context management with Redis
• Streaming responses
• Parallel execution
• Integration with Anthropic
• UI Component discovery

import json
import time
from typing import Generator
from anthropic import Anthropic
import agentstable

# Configuration
ANTHROPIC_API_KEY = "your_anthropic_api_key"
API_TOKEN = "your_api_token"
BASE_URL = "http://your_service_url/api"
COMPONENT_SERVICE_URL = "http://your_component_service_url"
REDIS_URL = "redis://username:password@host:port"

# Auth setup
auth = agentstable.BearerAuth(token=API_TOKEN)

# ---- Memory Setup with Redis ----
# Create a session with Redis storage
session = agentstable.create_session(
    session_id="vacation_planning_session",
    use_redis=True,
    redis_url=REDIS_URL
)

# Set initial context
session.set_context("preferred_destination", "Hawaii", "travel_flow")
session.set_context("trip_duration", 7, "travel_flow")
session.set_context("budget", 5000, "travel_flow")
print(f"Initial context: {json.dumps(session.get_all_context('travel_flow'), indent=2)}")

# ---- Search Functionality ----
# Create context-aware query
query = f"Plan a vacation to {session.get_context('preferred_destination', 'travel_flow')}"
print(f"Query: '{query}'")

# Search for matching flow
flow = agentstable.search(
    query=query,
    collection_id="vacation",
    base_url=BASE_URL,
    auth=auth
)
flow_id = flow.get("flow_id")
print(f"Found flow: {flow_id}")

# ---- Component Discovery ----
# Search for UI components that can display vacation data
component_query = "Display vacation destinations with images and details"
components = agentstable.search_components(
    query=component_query,
    base_url=COMPONENT_SERVICE_URL,
    limit=3
)

# Show discovered components
if isinstance(components, dict) and "results" in components:
    component_results = components.get("results", [])
else:
    component_results = components

print(f"Found {len(component_results)} matching UI components")
for i, component in enumerate(component_results, 1):
    print(f"Component {i}: {component.get('name')} - {component.get('description', '')[:100]}...")

# Format the best component for display
if component_results:
    best_component = component_results[0]
    formatted_component = agentstable.format_component_for_display(best_component)
    session.set_context("best_component", best_component.get('id'), "travel_flow")
    session.set_context("formatted_component", formatted_component, "travel_flow")
    print(f"Best component: {best_component.get('name')}")

# ---- Streaming Responses ----
# Set up Anthropic client and tools
client = Anthropic(api_key=ANTHROPIC_API_KEY)
tools = agentstable.get_anthropic_tools(flow)

# Define callback for streaming
def handle_stream_chunk(content_type, content):
    if content_type == "text":
        print(content, end="", flush=True)
    elif content_type == "tool_use":
        print(f"\n[Tool Call] {content}")
        # Store tool call in context
        try:
            tool_data = json.loads(content)
            session.set_context("last_tool_call", tool_data, "travel_flow")
        except:
            pass

# Stream response
print("\nStreaming response:")
messages = [{"role": "user", "content": query}]
complete_response = None

for chunk in agentstable.stream_anthropic(
    client=client,
    messages=messages,
    tools=tools,
    model="claude-3-sonnet-20240229",
    callback=handle_stream_chunk
):
    # Callback handles printing
    # We keep track of the full response for later use
    if isinstance(chunk, dict) and not isinstance(chunk, Generator):
        complete_response = chunk

# ---- Parallel Execution ----
# Extract actions from flow
flow_schema = None
for f in flow.get("schema", {}).get("flows", []):
    if f.get("id") == flow_id:
        flow_schema = f
        break

if flow_schema:
    actions = flow_schema.get("actions", [])
    action_ids = [a.get("id") for a in actions if a.get("id")]
    print(f"\nFlow has {len(action_ids)} actions: {action_ids}")

    # Get tool call arguments
    arguments = {}
    try:
        # From context (stored during streaming)
        tool_call = session.get_context("last_tool_call", "travel_flow")
        if tool_call and isinstance(tool_call, dict):
            arguments = tool_call.get("input", {})
    except:
        # Fallback arguments
        arguments = {
            "location": "Hawaii",
            "days": 7,
            "place_type": "attraction"
        }

    # Sequential execution
    print("\nExecuting actions sequentially...")
    sequential_start = time.time()
    result = agentstable.execute_anthropic(
        flow_response=flow,
        anthropic_response=complete_response,
        auth=auth,
        session=session
    )
    sequential_time = time.time() - sequential_start

    # Parallel execution
    print("\nExecuting actions in parallel...")
    parallel_start = time.time()
    parallel_results = agentstable.execute_actions_parallel(
        flow_response=flow,
        action_ids=action_ids,
        auth=auth,
        arguments=arguments,
        max_workers=len(action_ids),
        session=session
    )
    parallel_time = time.time() - parallel_start

    # Performance comparison
    speedup = sequential_time / parallel_time if parallel_time > 0 else 0
    print(f"\nSpeedup with parallel execution: {speedup:.2f}x")

# ---- Memory Persistence ----
# Show context after operations
print("\nFinal context:")
print(f"Number of items: {len(session.get_all_context('travel_flow'))}")
print(f"Context keys: {list(session.get_all_context('travel_flow').keys())}")

# Create a new session to demonstrate persistence
new_session = agentstable.create_session(
    session_id="vacation_planning_session",
    use_redis=True,
    redis_url=REDIS_URL
)

# Verify data persistence across sessions
print(f"\nData persisted across sessions: {len(new_session.get_all_context('travel_flow'))} items")
destination = new_session.get_context("preferred_destination", "travel_flow")
duration = new_session.get_context("trip_duration", "travel_flow")
print(f"Planning a vacation to {destination} for {duration} days")

Advanced Usage

Memory and Context Management

AgentStable SDK provides memory and context management capabilities to maintain state across API calls:

import agentstable

# Create a session to manage context
session = agentstable.create_session("my_session")

# Add context variables manually
session.set_context("user_id", "12345", flow_id="user_flow")
session.set_context("preferences", {"theme": "dark"}, flow_id="user_flow")

# Execute actions with the session
result = agentstable.execute_openai(
    flow, response, auth, session=session
)

# The session automatically stores results from previous actions
# and makes them available to subsequent calls
print(session.get_all_context("user_flow"))

# Access specific context variables
user_id = session.get_context("user_id", flow_id="user_flow")

Persistent Redis Storage

For persistent storage across restarts and processes, AgentStable supports Redis:

# Create a session with Redis as the storage backend
session = agentstable.create_session(
    session_id="persistent_session_id",
    use_redis=True,
    redis_url="redis://username:password@host:port"
)

# Or use the REDIS_URL environment variable
import os
os.environ["REDIS_URL"] = "redis://username:password@host:port"
session = agentstable.create_session(
    session_id="persistent_session_id",
    use_redis=True
)

# Use the session as normal - all data will be stored in Redis
session.set_context("user_data", {"name": "Alice"}, "user_flow")

# Data persists across sessions with the same ID
new_session = agentstable.create_session(
    session_id="persistent_session_id",
    use_redis=True
)
user_data = new_session.get_context("user_data", "user_flow")  # Returns {"name": "Alice"}

Session Features

• Persistence across calls: Context is maintained between different API calls
• Flow-specific context: Each flow can have its own isolated context
• Automatic context sharing: Results from previous actions are available to future actions
• Conversation history: Track the history of user-assistant interactions
• Redis storage: Optional persistent storage using Redis

Direct Search and Tool Generation

You can combine searching and tool generation in a single call:

# For OpenAI
openai_result = agentstable.search_and_generate_openai_tools(
    query="Create a product called Premium Access for $100",
    collection_id="your_collection_id"
)

# For Anthropic
anthropic_result = agentstable.search_and_generate_anthropic_tools(
    query="Create a product called Premium Access for $100",
    collection_id="your_collection_id"
)

UI Component Discovery

AgentStable SDK includes functionality to discover UI components using natural language queries. This helps AI agents recommend appropriate interface elements to display data retrieved through actions.

import agentstable

# Search for UI components that match a description
components = agentstable.search_components(
    query="Display a table of users with search functionality",
    base_url="http://your_component_service_url",
    limit=3
)

# Display the results
for component in components:
    print(f"Component: {component.get('name')}")
    print(f"Description: {component.get('description')}")
    print(f"Tags: {', '.join(component.get('tags', []))}")
    print("---")

# Format a component for display in documentation or UI
if components:
    formatted = agentstable.format_component_for_display(components[0])
    print(formatted)

Component Features

• Semantic search: Find components based on natural language descriptions
• Component metadata: Access details like name, description, tags, and implementation details
• Formatted display: Convert component data into human-readable documentation
• Integration with actions: Combine data from API actions with UI components
• Framework support: Components from various UI frameworks (React, Vue, etc.)

Different Authentication Methods

The SDK supports multiple authentication methods:

# Bearer token authentication
auth = agentstable.BearerAuth(token="your_token")

# API key authentication
auth = agentstable.ApiKeyAuth(api_key="your_api_key", header_name="X-API-Key")

# Basic authentication
auth = agentstable.BasicAuth(username="your_username", password="your_password")

# No authentication
auth = agentstable.NoAuth()

Get Available Collections and Schemas

# Get all collections
collections = agentstable.get_all_collections()

# Get all schemas in a collection
schemas = agentstable.get_all_schemas(collection_id="your_collection_id")

Advanced Usage Features

Memory and Context Management

The SDK provides a session-based memory system that persists across API calls:

from agentstable import create_session

# Create a new session (in-memory by default)
session = create_session(session_id="my_session_id")

# Add variables to the session context
session.set_context("user_name", "Alice", namespace="user_info")
session.set_context("preferred_language", "English", namespace="user_prefs")

# Retrieve context values
user_name = session.get_context("user_name", namespace="user_info")
all_prefs = session.get_all_context(namespace="user_prefs")

# Clear context in a specific namespace
session.clear_context(namespace="user_prefs")

Token Usage Tracking

The SDK provides automatic token usage tracking to help you monitor your LLM consumption:

from agentstable import get_usage_tracker

# Get the usage tracker
usage_tracker = get_usage_tracker()

# Get usage summary
summary = usage_tracker.get_usage_summary()
print(f"Total API calls: {summary['total_calls']}")
print(f"Total tokens: {summary['total_tokens']}")

Action Generator

The SDK includes an ActionGenerator that helps you create JSON schemas for actions from natural language:

from openai import OpenAI
from agentstable import OpenAIActionGenerator

# Create the OpenAI client
client = OpenAI(api_key="your-openai-api-key")

# Create the action generator
generator = OpenAIActionGenerator(client=client)

# Generate a schema from a natural language query
schema = generator.generate_from_query(
    query="Create an API for a weather service that gets weather and forecasts for cities"
)

# Convert to action service format
action_service_schema = generator.convert_to_action_service(
    schema=schema,
    collection_id="weather_service"
)

For more details, see the Action Generator documentation.

API Reference

Search Functions

• search(query, collection_id, base_url) - Search for a flow using natural language
• search_and_generate_tools(query, collection_id, base_url, provider) - Search and generate tools for any supported provider
• search_and_generate_openai_tools(query, collection_id, base_url) - Search and generate OpenAI tools
• search_and_generate_anthropic_tools(query, collection_id, base_url) - Search and generate Anthropic tools

Tool Generation

• get_tools(response) - Generate OpenAI tools from a flow response
• get_anthropic_tools(response) - Generate Anthropic tools from a flow response

Execution Functions

• execute(flow_response, llm_response, auth, base_url, provider, session) - Execute API calls based on LLM output
• execute_openai(flow_response, openai_response, auth, base_url, session) - Execute using OpenAI responses
• execute_anthropic(flow_response, anthropic_response, auth, base_url, session) - Execute using Anthropic responses
• execute_actions_parallel(flow_response, action_ids, auth, base_url, arguments, max_workers, session) - Execute multiple actions in parallel
• execute_flows_parallel(flow_responses, llm_responses, auth, base_url, provider, max_workers, session) - Execute multiple flows in parallel

Streaming Functions

• stream_anthropic(client, messages, tools, model, max_tokens, temperature, callback) - Stream responses from Anthropic
• stream_openai(client, messages, tools, model, temperature, callback) - Stream responses from OpenAI

Memory and Context Management

• create_session(session_id, use_redis, redis_url) - Create a new session for context management
• get_session() - Get the default session
• session.set_context(key, value, flow_id) - Store a value in session context
• session.get_context(key, flow_id, default) - Retrieve a value from session context
• session.get_all_context(flow_id) - Get all context for a flow
• session.clear_context(flow_id) - Clear context for a flow

Documentation

For complete documentation, visit the GitHub repository.

License

This project is licensed under the MIT License - see the LICENSE file for details.