Daita Logo

REST API Plugin

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

#Installation

bash
pip install aiohttp

#Quick Start

python
from daita import Agent
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 = Agent(
    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")

#Direct Usage

The plugin can be used directly without agents for programmatic access. For comprehensive HTTP client documentation, see the aiohttp docs. The main value of this plugin is agent integration - enabling LLMs to autonomously interact with REST APIs.

#Connection Parameters

python
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

python
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

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

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

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

python
from daita import Agent
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 = Agent(
    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:

ToolDescriptionParameters
http_getMake GET requestendpoint (required), params (object)
http_postMake POST requestendpoint (required), data (object)
http_putMake PUT requestendpoint (required), data (object)
http_deleteMake DELETE requestendpoint (required), params (object)

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

#Tool Usage Example

python
from daita import Agent
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 = Agent(
    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

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

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

python
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

IssueSolution
aiohttp not installedpip install aiohttp
Connection failedCheck base URL, internet connectivity, firewall
HTTP 401Verify API key and authentication header format
HTTP 429Implement retry with exponential backoff
HTTP 500Check API server status, verify request payload
TimeoutIncrease timeout parameter, check network

#Next Steps