REST API Plugin

Simple HTTP client for API integrations with authentication support and file transfers. Built on aiohttp.

Installation

pip install aiohttp

Quick Start

Direct Usage (Scripts)

from daita.plugins import rest

# Direct usage in scripts
async with rest(base_url="https://api.example.com") as api:
    data = await api.get("/users")
    print(data)

Agent Integration (Recommended)

from daita import SubstrateAgent
from daita.plugins import rest

# Create plugin
api = rest(base_url="https://api.example.com", api_key="your-key")

# Agent uses REST tools autonomously
agent = SubstrateAgent(
    name="API Agent",
    prompt="You are an API integration specialist. Help users interact with REST APIs.",
    tools=[api]
)

await agent.start()
result = await agent.run("Get all users from the API")

Connection Parameters

rest(
    base_url: str,
    api_key: Optional[str] = None,
    auth_header: str = "Authorization",
    auth_prefix: str = "Bearer",
    timeout: int = 30,
    **kwargs
)

Parameters

• base_url (str): Base URL for all requests (required)
• api_key (str): API key for authentication (optional)
• auth_header (str): Header name for authentication (default: "Authorization")
• auth_prefix (str): Prefix for auth value (default: "Bearer")
• timeout (int): Request timeout in seconds (default: 30)
• **kwargs: Additional configuration (e.g., custom headers)

HTTP Methods

async with rest(base_url="https://api.example.com") as api:
    # GET
    users = await api.get("/users")
    filtered = await api.get("/users", params={"status": "active", "page": 1})

    # POST
    new_user = await api.post("/users", json_data={
        "name": "John Doe",
        "email": "john@example.com"
    })

    # PUT
    updated = await api.put(f"/users/{user_id}", json_data={"name": "Jane"})

    # PATCH
    patched = await api.patch(f"/users/{user_id}", json_data={"status": "inactive"})

    # DELETE
    result = await api.delete(f"/users/{user_id}")

Authentication

# Bearer token (default)
async with rest(
    base_url="https://api.example.com",
    api_key="your-api-key"
) as api:
    # Adds: Authorization: Bearer your-api-key
    data = await api.get("/protected")

# Custom header
async with rest(
    base_url="https://api.example.com",
    api_key="your-token",
    auth_header="X-API-Key",
    auth_prefix=""  # No prefix
) as api:
    # Adds: X-API-Key: your-token
    data = await api.get("/users")

File Operations

async with rest(base_url="https://api.example.com") as api:
    # Upload
    result = await api.upload_file(
        endpoint="/upload",
        file_path="/path/to/document.pdf",
        field_name="document"
    )

    # Download
    local_path = await api.download_file(
        endpoint="/files/report.pdf",
        save_path="/local/downloads/report.pdf"
    )

Responses are automatically parsed (JSON, text, or binary) based on content type.

Using with Agents

Direct HTTP Operations (Scripts)

For scripts that don't need agent capabilities:

from daita.plugins import rest

async with rest(
    base_url="https://api.example.com",
    api_key="your-api-key"
) as api:
    # Direct API calls
    response = await api.get("/users", params={"limit": 100})
    user_data = await api.get(f"/users/{user_id}")

    print(f"Users: {response}")
    print(f"User data: {user_data}")

Tool-Based Integration (Recommended)

REST plugin exposes HTTP operations as tools that agents can use autonomously:

from daita import SubstrateAgent
from daita.plugins import rest
import os

# Create REST API plugin
api = rest(
    base_url="https://api.example.com",
    api_key=os.getenv("API_KEY")
)

# Pass plugin to agent - agent can now use HTTP tools autonomously
agent = SubstrateAgent(
    name="API Agent",
    prompt="You are an API integration specialist. Help users interact with REST APIs.",
    llm_provider="openai",
    model="gpt-4",
    tools=[api]
)

await agent.start()

# Agent autonomously uses HTTP tools to answer questions
result = await agent.run("Fetch all active users from the API")

# The agent will autonomously:
# 1. Use http_get tool to fetch data
# 2. Analyze the response
# 3. Present results in natural language

await agent.stop()

Available Tools

The REST plugin exposes these tools to LLM agents:

Tool	Description	Parameters
http_get	Make GET request	endpoint (required), params (object)
http_post	Make POST request	endpoint (required), data (object)
http_put	Make PUT request	endpoint (required), data (object)
http_delete	Make DELETE request	endpoint (required), params (object)

Tool Categories: api Tool Source: plugin Base URL: Configured at plugin initialization

Tool Usage Example

from daita import SubstrateAgent
from daita.plugins import rest

# Setup REST API with tool integration
api = rest(
    base_url="https://api.github.com",
    auth_header="Authorization",
    auth_prefix="Bearer",
    api_key="github_token"
)

agent = SubstrateAgent(
    name="GitHub Assistant",
    prompt="You are a GitHub assistant. Help users interact with the GitHub API.",
    llm_provider="openai",
    model="gpt-4",
    tools=[api]
)

await agent.start()

# Natural language command - agent uses tools autonomously
result = await agent.run("""
Get information about user 'octocat':
1. Fetch user profile
2. Get their repositories
3. Summarize activity
""")

# Agent orchestrates HTTP tool calls to fulfill the request
print(result)
await agent.stop()

Error Handling

try:
    async with rest(base_url="https://api.example.com") as api:
        data = await api.get("/users")
except RuntimeError as e:
    if "aiohttp not installed" in str(e):
        print("Install aiohttp: pip install aiohttp")
    elif "HTTP 401" in str(e):
        print("Authentication failed")
    elif "HTTP 429" in str(e):
        print("Rate limit exceeded")

Best Practices

Security:

• Store API keys in environment variables, never hardcode
• Use HTTPS for production endpoints
• Validate base URLs before use

Performance:

• Reuse connections within handlers (use single async with block)
• Batch requests when API supports it
• Set appropriate timeouts to prevent hanging

Request Configuration:

• Use query parameters dict instead of hardcoded query strings
• Set timeouts based on API response times
• Handle errors gracefully with context

Common Patterns

Pagination:

async def fetch_all_pages(agent):
    all_data, page = [], 1
    async with rest(base_url="https://api.example.com") as api:
        while True:
            response = await api.get("/data", params={"page": page, "limit": 100})
            if not response.get("data"):
                break
            all_data.extend(response["data"])
            if not response.get("has_more"):
                break
            page += 1
    return all_data

Retry with backoff:

import asyncio

async def fetch_with_retry(agent, endpoint, max_retries=3):
    async with rest(base_url="https://api.example.com") as api:
        for attempt in range(max_retries):
            try:
                return await api.get(endpoint)
            except RuntimeError as e:
                if "HTTP 429" in str(e) and attempt < max_retries - 1:
                    await asyncio.sleep(2 ** attempt)
                else:
                    raise

Troubleshooting

Issue	Solution
aiohttp not installed	pip install aiohttp
Connection failed	Check base URL, internet connectivity, firewall
HTTP 401	Verify API key and authentication header format
HTTP 429	Implement retry with exponential backoff
HTTP 500	Check API server status, verify request payload
Timeout	Increase timeout parameter, check network

Next Steps

• S3 Plugin - Cloud storage operations
• Slack Plugin - Team notifications
• PostgreSQL Plugin - Database integration
• Plugin Overview - All available plugins