Python SDK Reference

The official Python SDK for Agent OTP. Full type hints, async support, and comprehensive error handling.

Installation

pip install agent-otp

Or with poetry:

poetry add agent-otp

Requirements

• Python 3.9 or later
• Works with both sync and async code

Quick Start

from agent_otp import AgentOTPClient

# Initialize the client
client = AgentOTPClient(api_key="ak_live_xxxx")

# Request permission
permission = client.request_permission(
    action="email.send",
    resource="email:client@example.com",
    scope={
        "max_emails": 1,
        "allowed_recipients": ["client@example.com"],
    },
    context={
        "reason": "Sending monthly invoice",
    },
    wait_for_approval=True,
    timeout=60,
)

if permission.status == "approved":
    print(f"Token: {permission.token}")

    # Use the token
    send_email(recipient="client@example.com", otp_token=permission.token)

    # Mark token as used
    client.use_token(permission.id, permission.token)

Async Support

The Python SDK supports async/await for use with asyncio:

import asyncio
from agent_otp import AsyncAgentOTPClient

async def main():
    client = AsyncAgentOTPClient(api_key="ak_live_xxxx")

    permission = await client.request_permission(
        action="file.read",
        scope={"max_size": 1048576},
        wait_for_approval=True,
    )

    if permission.status == "approved":
        # Use the token
        await process_file(permission.token)
        await client.use_token(permission.id, permission.token)

asyncio.run(main())

Client Configuration

from agent_otp import AgentOTPClient

client = AgentOTPClient(
    api_key="ak_live_xxxx",
    base_url="https://api.agentotp.com",  # Optional
    timeout=30.0,                          # Request timeout in seconds
    retries=3,                             # Number of retry attempts
    debug=False,                           # Enable debug logging
)

Permission Request

Parameters

Error Handling

from agent_otp import (
    AgentOTPClient,
    AgentOTPError,
    AuthenticationError,
    ValidationError,
    RateLimitError,
    NetworkError,
)

client = AgentOTPClient(api_key="ak_live_xxxx")

try:
    permission = client.request_permission(
        action="email.send",
        scope={"max_emails": 1},
    )
except AuthenticationError as e:
    print(f"Invalid API key: {e}")
except ValidationError as e:
    print(f"Invalid request: {e.details}")
except RateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after}s")
except NetworkError as e:
    print(f"Network error: {e}")
except AgentOTPError as e:
    print(f"API error: {e.code} - {e.message}")

Context Manager

Use the client as a context manager to ensure proper cleanup:

from agent_otp import AgentOTPClient

with AgentOTPClient(api_key="ak_live_xxxx") as client:
    permission = client.request_permission(
        action="file.write",
        scope={"max_size": 1048576},
    )

    # Client is automatically closed when exiting the context

Type Hints

The SDK is fully typed for excellent IDE support:

from agent_otp import (
    AgentOTPClient,
    PermissionRequest,
    PermissionResponse,
    PermissionStatus,
)

def process_with_permission(
    client: AgentOTPClient,
    action: str,
) -> PermissionResponse:
    request: PermissionRequest = {
        "action": action,
        "scope": {"max_count": 1},
    }

    response: PermissionResponse = client.request_permission(**request)

    if response.status == PermissionStatus.APPROVED:
        return response

    raise ValueError(f"Permission {response.status}")

Integration Examples

LangChain

from langchain.tools import BaseTool
from agent_otp import AgentOTPClient

class OTPProtectedTool(BaseTool):
    name = "email_sender"
    description = "Send emails with OTP protection"

    def __init__(self):
        self.otp_client = AgentOTPClient(api_key="ak_live_xxxx")

    def _run(self, recipient: str, subject: str, body: str) -> str:
        permission = self.otp_client.request_permission(
            action="email.send",
            resource=f"email:{recipient}",
            scope={"max_emails": 1},
            wait_for_approval=True,
        )

        if permission.status != "approved":
            return f"Permission denied: {permission.reason}"

        # Send email using the token
        result = send_email(recipient, subject, body, permission.token)
        self.otp_client.use_token(permission.id, permission.token)

        return "Email sent successfully"

See Also

• Quick Start Guide
• Error Handling Guide
• LangChain Integration