Implementing Agent Protocol

Build a Compliant Agent in Minutes

For Agent Developers

Option 1: Use the Official SDK (Recommended)

The Agent Protocol SDK handles all API infrastructure, letting you focus solely on your agent’s logic.

Python Implementation:

python

from agent_protocol import Agent, Task, Step

# Define your agent logic
async def task_handler(task: Task) -> None:
    # Your agent's task execution logic
    print(f"Executing task: {task.input}")
    
async def step_handler(step: Step) -> Step:
    # Your agent's step execution logic
    step.output = "Step completed successfully"
    return step

# Create agent with the protocol
agent = Agent(task_handler=task_handler, step_handler=step_handler)

# Start the server
if __name__ == "__main__":
    agent.start(port=8000)

Installation:

bash

pip install agent-protocol

JavaScript/TypeScript Implementation:

typescript

import { Agent } from 'agent-protocol';

const agent = new Agent({
  taskHandler: async (task) => {
    // Your task logic
  },
  stepHandler: async (step) => {
    // Your step logic
    return step;
  }
});

agent.start(8000);

Installation:

bash

npm install agent-protocol

Option 2: Implement the API Manually

If you prefer full control, you can implement the REST API directly using any web framework.

Example with FastAPI (Python):

python

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import uuid

app = FastAPI()

# Data models
class TaskInput(BaseModel):
    input: str
    additional_input: Optional[dict] = None

class Task(BaseModel):
    task_id: str
    input: str
    artifacts: List[str] = []

class Step(BaseModel):
    step_id: str
    task_id: str
    name: str
    status: str
    output: str
    is_last: bool = False

# In-memory storage (use database in production)
tasks = {}
steps = {}

@app.post("/ap/v1/agent/tasks", response_model=Task)
async def create_task(task_input: TaskInput):
    task_id = str(uuid.uuid4())
    task = Task(
        task_id=task_id,
        input=task_input.input,
        artifacts=[]
    )
    tasks[task_id] = task
    return task

@app.post("/ap/v1/agent/tasks/{task_id}/steps", response_model=Step)
async def execute_step(task_id: str):
    if task_id not in tasks:
        raise HTTPException(status_code=404, detail="Task not found")
    
    # Your agent logic here
    step_id = str(uuid.uuid4())
    step = Step(
        step_id=step_id,
        task_id=task_id,
        name="Agent Step",
        status="completed",
        output="Step result",
        is_last=False
    )
    steps[step_id] = step
    return step

# Add other endpoints...

View Complete Implementation Examples →

For Agent Users (Integration)

Using the Client Library:

python

from agent_protocol_client import AgentApi, ApiClient, Configuration

# Configure client
config = Configuration(host="http://localhost:8000")
client = ApiClient(configuration=config)
agent = AgentApi(client)

# Create a task
task_input = {"input": "Write a Python script to analyze CSV data"}
task = agent.create_agent_task(body=task_input)
print(f"Task created: {task.task_id}")

# Execute steps until completion
while True:
    step = agent.execute_agent_task_step(task_id=task.task_id)
    print(f"Step output: {step.output}")
    
    if step.is_last:
        break

print("Task completed!")

Client Libraries Available:

• Python: pip install agent-protocol-client
• JavaScript/TypeScript: npm install agent-protocol-client
• Go: Coming soon
• Rust: Coming soon

Testing Your Implementation

1. Start your agent:

bash

python your_agent.py
# Agent running on http://localhost:8000

2. Test with curl:

bash

# Create a task
curl -X POST http://localhost:8000/ap/v1/agent/tasks \
  -H "Content-Type: application/json" \
  -d '{"input": "Hello, agent!"}'

# Execute a step
curl -X POST http://localhost:8000/ap/v1/agent/tasks/TASK_ID/steps

3. Use benchmarking tools:

bash

# Install AutoGPT's benchmark
pip install agbenchmark

# Run against your agent
agbenchmark --test=your_agent_url

Best Practices

1. Stateless Design

• Store task/step data in a database, not in memory
• Enable horizontal scaling

2. Async Operations

• Use async/await for step execution
• Don’t block the API during long-running operations

3. Error Handling

• Return meaningful error messages
• Use appropriate HTTP status codes

4. Artifact Management

• Store large files externally (S3, etc.)
• Return URIs instead of embedding data

5. Documentation

• Document your agent’s capabilities
• Provide usage examples