AMP
GitHub
SDK Reference

Python SDK

Official Python SDK for AMP.

Installation

pip install amp-protocol

Initialisation

from amp_protocol import AMP
import os

amp = AMP(
    api_key=os.getenv("AMP_API_KEY"),

    # Optional configuration
    base_url="https://api.amp-protocol.dev/v1",  # Default
    timeout=5.0,                                   # Request timeout (seconds)
    max_retries=3,                                 # Auto-retry failed requests
    debug=False                                    # Enable debug logging
)

Methods

get_context()

context = await amp.get_context(
    user_id="user_123",
    task="build a feature",
    complexity="medium",      # Optional
    task_type="coding",       # Optional
    metadata={}               # Optional
)

report_outcome()

await amp.report_outcome(
    request_id=context.request_id,
    started=True,
    completed=True,
    time_to_start=45,         # Optional
    flow_state=True,          # Optional
    satisfaction=0.9          # Optional
)

update_profile()

profile = await amp.update_profile(
    user_id="user_123",
    preferred_framing="achievement",
    communication_style="brief_directive"
)

Type Hints

from amp_protocol import (
    AMP,
    MotivationContext,
    FramingType,
    CommunicationStyle
)

# Fully typed with Pydantic!
context: MotivationContext = await amp.get_context(
    user_id="user_123",
    task="test"
)

Error Handling

from amp_protocol.exceptions import (
    AMPError,
    RateLimitError,
    UnauthorizedError
)

try:
    context = await amp.get_context(user_id=user_id, task=task)
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except UnauthorizedError as e:
    print("Invalid API key")
except AMPError as e:
    print(f"AMP Error: {e.code} - {e.message}")

Event Hooks

# Register event listeners
@amp.on("request")
async def on_request(params):
    print(f"AMP Request: {params}")

@amp.on("response")
async def on_response(context, duration):
    print(f"AMP responded in {duration}ms")

@amp.on("error")
async def on_error(error):
    print(f"AMP Error: {error}")

Framework Examples

FastAPI

main.py
from fastapi import FastAPI
from amp_protocol import AMP
import os

app = FastAPI()
amp = AMP(api_key=os.getenv("AMP_API_KEY"))

@app.post("/api/agent")
async def handle_query(request: dict):
    context = await amp.get_context(
        user_id=request["userId"],
        task=request["query"]
    )

    return {"context": context.dict()}

Flask

app.py
from flask import Flask, request, jsonify
from amp_protocol import AMP
import asyncio
import os

app = Flask(__name__)
amp = AMP(api_key=os.getenv("AMP_API_KEY"))

@app.route("/api/agent", methods=["POST"])
def handle_query():
    data = request.json

    context = asyncio.run(amp.get_context(
        user_id=data["userId"],
        task=data["query"]
    ))

    return jsonify({"context": context.dict()})

Django

views.py
from django.http import JsonResponse
from amp_protocol import AMP
import os

amp = AMP(api_key=os.getenv("AMP_API_KEY"))

async def agent_view(request):
    import json
    data = json.loads(request.body)

    context = await amp.get_context(
        user_id=data["userId"],
        task=data["query"]
    )

    return JsonResponse({"context": context.dict()})

💡 Async Support: The SDK fully supports both async/await and synchronous usage patterns.

Synchronous Usage

# Synchronous client
from amp_protocol import AMPSync

amp_sync = AMPSync(api_key=os.getenv("AMP_API_KEY"))

# Use without async/await
context = amp_sync.get_context(user_id="user_123", task="test")
amp_sync.report_outcome(request_id=context.request_id, started=True)

See Also