Request Signing Guide

from adcp.signing import generate_signing_keypair

# CLI equivalence: adcp-keygen --alg ed25519 --purpose request-signing --kid my-agent-2026
pem_bytes, public_jwk = generate_signing_keypair(
    alg="ed25519",
    purpose="request-signing",
    kid="my-agent-2026",
)
# pem_bytes — write to disk with mode 0600 and O_EXCL, or pass to a secret manager.
# public_jwk — publish in your JWKS endpoint. Already includes kid, use, key_ops, and adcp_use.

import "github.com/adcontextprotocol/adcp-go/adcp/signing"

res, err := signing.GenerateKeyForProfile(
    signing.AlgEd25519,
    "my-agent-2026",
    signing.ProfileRequestSigning,
)
if err != nil { /* handle */ }
// res.PrivateKeyPEM — write to disk with mode 0600, or pass to a secret manager.
// res.PublicJWK    — serialize and publish in your JWKS endpoint. AdCP-required
//                    fields (kid, kty, crv, alg, use, key_ops, adcp_use) are set.

from adcp import ADCPClient
from adcp.signing import SigningConfig, load_private_key_pem

private_key = load_private_key_pem(open("private-key.pem", "rb").read())

client = ADCPClient(
    base_url="https://seller.example.com/mcp",
    signing=SigningConfig(
        key_id="my-agent-2026",
        alg="ed25519",
        private_key=private_key,
        cover_content_digest=True,
    ),
)
# Every request the client sends carries Signature, Signature-Input, and
# Content-Digest headers.

from adcp.signing import sign_request

signed = sign_request(
    method="POST",
    url="https://seller.example.com/mcp",
    headers={"Content-Type": "application/json"},
    body=request_body_bytes,
    private_key=private_key,
    key_id="my-agent-2026",
    alg="ed25519",
    cover_content_digest=True,
)
# signed.as_dict() returns the headers to attach to the outgoing request.

import (
    "net/http"
    "github.com/adcontextprotocol/adcp-go/adcp/signing"
)

priv, _, _ := signing.LoadPrivateKey(pemBytes)
signer, _ := signing.NewSigner(signing.SignerOptions{
    KeyID:      "my-agent-2026",
    Algorithm:  signing.AlgEd25519,
    PrivateKey: priv,
})

client := &http.Client{
    Transport: signer.RoundTripper(http.DefaultTransport, true /* cover content-digest */),
}

req, _ := http.NewRequest("POST", "https://seller.example.com/mcp", body)
req.Header.Set("Content-Type", "application/json")
resp, _ := client.Do(req)
// Signature, Signature-Input, and Content-Digest are added by the transport.

from fastapi import FastAPI, Request, HTTPException
from adcp.signing import (
    VerifyOptions,
    VerifierCapability,
    CachingJwksResolver,
    InMemoryReplayStore,
    StaticRevocationChecker,
)
from adcp.signing.middleware import (
    verify_starlette_request,
    unauthorized_response_headers,
)
from adcp.signing.errors import SignatureVerificationError

app = FastAPI()

verify_options = VerifyOptions(
    capability=VerifierCapability(
        supported=True,
        covers_content_digest="required",
        required_for={"create_media_buy", "update_media_buy"},
    ),
    jwks_resolver=CachingJwksResolver(),
    replay_store=InMemoryReplayStore(),
    revocation_checker=StaticRevocationChecker(set()),
)

@app.post("/mcp")
async def mcp(request: Request):
    try:
        signer = await verify_starlette_request(request, options=verify_options)
    except SignatureVerificationError as exc:
        raise HTTPException(
            status_code=401,
            detail=exc.code,
            headers=unauthorized_response_headers(exc),
        )
    # signer.key_id, signer.agent_url, signer.verified_at available for audit.
    body = await request.body()
    return await handle_mcp(body, signer)

import (
    "net/http"
    "github.com/adcontextprotocol/adcp-go/adcp/signing"
)

resolver := signing.NewCachingJWKSResolver()
replay := signing.NewMemoryReplayStore(0 /* default cap */)
revocation := signing.NewStaticRevocationSource(nil)

mw := signing.Middleware(signing.MiddlewareOptions{
    Resolver:            resolver,
    Replay:              replay,
    Revocation:          revocation,
    OperationResolver:   signing.DefaultOperationResolver, // /adcp/<op>
    ContentDigestPolicy: signing.DigestRequired,
    RequiredFor:         []string{"create_media_buy", "update_media_buy"},
})

http.Handle("/mcp", mw(handler))

// Inside handler:
func handler(w http.ResponseWriter, r *http.Request) {
    v := signing.VerifiedSignerFromContext(r.Context())
    if v == nil {
        // Operation not in RequiredFor and request was unsigned — proceed
        // with bearer auth or whatever fallback you've configured.
    }
    // v.KeyID, v.AgentURL, v.VerifiedAt, v.Algorithm — available for audit.
}

from adcp.signing import (
    WebhookVerifyOptions,
    BrandJsonJwksResolver,
    InMemoryReplayStore,
    verify_webhook_signature,
)
from adcp.signing.errors import SignatureVerificationError

webhook_options = WebhookVerifyOptions(
    jwks_resolver=BrandJsonJwksResolver(),
    replay_store=InMemoryReplayStore(),
)

@app.post("/webhook")
async def webhook(request: Request):
    body = await request.body()
    try:
        sender = verify_webhook_signature(
            method=request.method,
            url=str(request.url),
            headers=dict(request.headers),
            body=body,
            options=webhook_options,
        )
    except SignatureVerificationError:
        raise HTTPException(status_code=401, detail="invalid webhook signature")
    # sender.key_id, sender.agent_url available for audit; process the webhook.

import (
    "github.com/adcontextprotocol/adcp-go/adcp/signing"
)

// Mount the same Middleware on your webhook receiver, but configure it for
// the webhook profile — adcp_use="webhook-signing", Content-Digest required,
// no required_for gating (webhooks always carry signatures).
webhookMW := signing.Middleware(signing.MiddlewareOptions{
    Resolver:            signing.NewBrandJSONJWKSResolver(),
    Replay:              signing.NewMemoryReplayStore(0),
    Revocation:          signing.NewStaticRevocationSource(nil),
    Profile:             signing.ProfileWebhookSigning,
    ContentDigestPolicy: signing.DigestRequired,
})

http.Handle("/webhook", webhookMW(webhookHandler))

func webhookHandler(w http.ResponseWriter, r *http.Request) {
    sender := signing.VerifiedSignerFromContext(r.Context())
    // sender.KeyID, sender.AgentURL — process the verified webhook.
}

from adcp.signing import sign_webhook, load_private_key_pem
import httpx, json

private_key = load_private_key_pem(open("webhook-private-key.pem", "rb").read())

async def post_webhook(url: str, payload: dict) -> None:
    body = json.dumps(payload).encode("utf-8")
    headers = {"Content-Type": "application/json"}
    signed = sign_webhook(
        method="POST",
        url=url,
        headers=headers,
        body=body,
        private_key=private_key,
        key_id="my-seller-webhook-2026",
        alg="ed25519",
    )
    headers.update(signed.as_dict())  # adds Signature, Signature-Input, Content-Digest
    async with httpx.AsyncClient() as client:
        await client.post(url, content=body, headers=headers)

priv, _, _ := signing.LoadPrivateKey(webhookPemBytes)
webhookSigner, _ := signing.NewSigner(signing.SignerOptions{
    KeyID:      "my-seller-webhook-2026",
    Algorithm:  signing.AlgEd25519,
    PrivateKey: priv,
    Profile:    signing.ProfileWebhookSigning,
})

webhookClient := &http.Client{
    Transport: webhookSigner.RoundTripper(http.DefaultTransport, true /* always cover content-digest for webhooks */),
}
// Use webhookClient.Post / .Do to deliver webhooks; signatures are added automatically.

from adcp.server.responses import capabilities_response

class MySeller(ADCPHandler):
    async def get_adcp_capabilities(self, params, context=None):
        return capabilities_response(
            ["media_buy"],
            request_signing={
                "supported": True,
                "required_for": ["create_media_buy", "update_media_buy"],
                "supported_for": ["sync_creatives", "sync_audiences"],
                "covers_content_digest": "either",
            },
        )

// In your get_adcp_capabilities handler, set the request_signing block on
// the response builder:
return adcp.CapabilitiesResponse(adcp.CapabilitiesData{
    SupportedProtocols: []string{"media-buy"},
    RequestSigning: &adcp.RequestSigningCapability{
        Supported:           true,
        RequiredFor:         []string{"create_media_buy", "update_media_buy"},
        SupportedFor:        []string{"sync_creatives", "sync_audiences"},
        CoversContentDigest: "either",
    },
}), nil