agentstable 0.1.4
A Python SDK for integrating with the AgentStable Search Action Service
Requires Python: >=3.8

# 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

```bash
pip install agentstable
```

## Quick Start

### Using with OpenAI

```python
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

```python
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

```python
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:

```python
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:

```python
# 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:

```python
# 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.

```python
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:

```python
# 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

```python
# 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:

```python
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:

```python
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:

```python
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](docs/action_generator.md).

## 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](https://github.com/clayton-dcruze/agentstable-sdk).

## License

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