Agent Endpoints

Expose custom HTTP API endpoints for your agent using the @http decorator. Endpoints are mounted under the agent’s base path and are served by the same FastAPI app used for chat completions.

Simple, declarative decorator: @http("/path", method="get|post", scope="...")

Path parameters and query strings supported

Scope-based access control (all, owner, admin)

Plays nicely with skills, tools, and hooks

Basic Usage

Define an endpoint and attach it to your agent via capabilities (auto-registration):

from webagents import BaseAgent, http

@http("/status", method="get")
def get_status() -> dict:
    return {"status": "healthy"}

agent = BaseAgent(
    name="assistant",
    model="openai/gpt-4o-mini",
    capabilities=[get_status]
)

Serve it (same as in Quickstart):

from webagents.server.core.app import create_server
import uvicorn

server = create_server(agents=[agent])
uvicorn.run(server.app, host="0.0.0.0", port=8000)

Available at:

GET /assistant/status

Methods, Path and Query

from webagents import http

# GET collection
@http("/users", method="get")
def list_users() -> dict:
    return {"users": ["alice", "bob", "charlie"]}

# POST create with JSON body
@http("/users", method="post")
def create_user(data: dict) -> dict:
    return {"created": data.get("name"), "id": "user_123"}

# GET item with path param and optional query param
@http("/users/{user_id}", method="get")
def get_user(user_id: str, include_details: bool = False) -> dict:
    user = {"id": user_id, "name": f"User {user_id}"}
    if include_details:
        user["details"] = "Extended info"
    return user

Example requests:

# List users
curl http://localhost:8000/assistant/users

# Create user
curl -X POST http://localhost:8000/assistant/users \
  -H "Content-Type: application/json" \
  -d '{"name": "dana"}'

# Get user with query param
curl "http://localhost:8000/assistant/users/42?include_details=true"

# Missing or wrong Content-Type
curl -X POST http://localhost:8000/assistant/users -d '{"name":"dana"}'
# -> 415 Unsupported Media Type

# Wrong method
curl -X GET http://localhost:8000/assistant/users -H "Content-Type: application/json" -d '{}'
# -> 405 Method Not Allowed

# Unauthorized scope (example)
curl http://localhost:8000/assistant/admin/metrics
# -> 403 Forbidden

Capability Discovery

Use provides= to declare what capability an endpoint provides:

@http("/export/pdf", method="post", provides="pdf_export")
def export_pdf(data: dict) -> bytes:
    """Export data as PDF."""
    return generate_pdf(data)

@http("/api/search", method="get", provides="search_api")
def search(query: str) -> dict:
    """Search API endpoint."""
    return {"results": perform_search(query)}

The provides value is included in the agent's capabilities for discovery.

Access Control (Scopes)

Use scope to restrict who can call an endpoint:

@http("/public", method="get", scope="all")
def public_endpoint() -> dict:
    return {"message": "Public data"}

@http("/owner-info", method="get", scope="owner")
def owner_endpoint() -> dict:
    return {"private": "owner data"}

@http("/admin/metrics", method="get", scope="admin")
def admin_metrics() -> dict:
    return {"rps": 100, "error_rate": 0.001}

WebSocket Endpoints

For bidirectional real-time communication, use the @websocket decorator:

from webagents import BaseAgent, websocket

@websocket("/stream")
async def my_websocket(ws) -> None:
    """Bidirectional WebSocket handler"""
    await ws.accept()
    try:
        async for message in ws.iter_json():
            # Process incoming message
            response = await process(message)
            await ws.send_json(response)
    except WebSocketDisconnect:
        pass

agent = BaseAgent(
    name="assistant",
    model="openai/gpt-4o-mini",
    capabilities=[my_websocket]
)

Available at:

WS /assistant/stream

WebSocket with LLM Streaming

Combine WebSocket with execute_handoff() in a skill:

from webagents.agents.skills.base import Skill
from webagents.agents.tools.decorators import websocket

class StreamingSkill(Skill):
    @websocket("/chat")
    async def chat_stream(self, ws) -> None:
        await ws.accept()
        
        async for msg in ws.iter_json():
            messages = msg.get("messages", [])
            
            # Stream LLM response through WebSocket
            async for chunk in self.execute_handoff(messages):
                await ws.send_json(chunk)

SSE Streaming (Server-Sent Events)

Return an AsyncGenerator from an @http handler to stream as SSE:

from webagents import http
from typing import AsyncGenerator

@http("/events", method="get")
async def stream_events() -> AsyncGenerator[str, None]:
    """SSE streaming endpoint"""
    for i in range(5):
        yield f"data: {{\"count\": {i}}}\n\n"
        await asyncio.sleep(1)
    yield "data: [DONE]\n\n"

The server automatically sets SSE headers:

Content-Type: text/event-stream

Cache-Control: no-cache

Connection: keep-alive

Tips

Keep one responsibility per endpoint (CRUD-style patterns work well)

Prefer get for retrieval, post for creation/processing

Validate inputs inside handlers; return JSON-serializable data

TypeScript Endpoints

HTTP Endpoints

Use the @http decorator in a Skill class to register HTTP endpoints:

import { Skill } from 'webagents/core/skill';
import { http } from 'webagents/core/decorators';
import type { Context } from 'webagents/core/types';

class APISkill extends Skill {
  @http({ path: '/status', method: 'GET' })
  async getStatus(request: Request, context: Context): Promise<Response> {
    return new Response(JSON.stringify({ status: 'healthy' }), {
      headers: { 'Content-Type': 'application/json' },
    });
  }

  @http({ path: '/process', method: 'POST' })
  async processData(request: Request, context: Context): Promise<Response> {
    const body = await request.json();
    const result = { processed: true, input: body };
    return new Response(JSON.stringify(result), {
      headers: { 'Content-Type': 'application/json' },
    });
  }
}

WebSocket Endpoints

Use the @websocket decorator:

import { Skill } from 'webagents/core/skill';
import { websocket } from 'webagents/core/decorators';
import type { Context } from 'webagents/core/types';

class StreamSkill extends Skill {
  @websocket({ path: '/stream' })
  handleStream(ws: WebSocket, context: Context): void {
    ws.onmessage = async (ev) => {
      const data = JSON.parse(ev.data as string);
      ws.send(JSON.stringify({ echo: data }));
    };
  }
}

Auto-Registration via Transport Skills

Transport skills register endpoints automatically when added to an agent. No manual endpoint registration is needed:

import { BaseAgent } from 'webagents/core/agent';
import { CompletionsTransportSkill } from 'webagents/skills/transport/completions/skill';
import { A2ATransportSkill } from 'webagents/skills/transport/a2a/skill';
import { UAMPTransportSkill } from 'webagents/skills/transport/uamp/skill';

const agent = new BaseAgent({
  name: 'my-agent',
  skills: [
    new CompletionsTransportSkill(),  // registers POST /v1/chat/completions, GET /v1/models
    new A2ATransportSkill(),          // registers POST /a2a, GET /.well-known/agent.json
    new UAMPTransportSkill(),         // registers WS /uamp
  ],
});

// Endpoints are now accessible via agent.getHttpHandler() and agent.getWebSocketHandler()
// Servers (node.ts, multi.ts) mount them automatically.

Agent Endpoints

Basic Usage

Methods, Path and Query

Capability Discovery

Access Control (Scopes)

WebSocket Endpoints

WebSocket with LLM Streaming

SSE Streaming (Server-Sent Events)

Tips

TypeScript Endpoints

HTTP Endpoints

WebSocket Endpoints

Auto-Registration via Transport Skills

See Also

On this page