test/simple_main.py

import os
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI(title="HF A2A Agent")

def hello_agent(message: str):
    return {"text": f"Hello from A2A on HF Spaces! You said: {message}"}

AGENT_CARD = {
    "protocolVersion": "0.3.0",
    "name": "Hello World Agent",
    "description": "A simple A2A agent that responds with 'Hello World' to any request",
    "url": "https://{HOST}/a2a",
    "preferredTransport": "HTTP+JSON",
    "version": "1.0.0",
    "capabilities": {
        "streaming": False,
        "pushNotifications": False,
        "stateTransitionHistory": False
    },
    "defaultInputModes": ["text/plain", "application/json"],
    "defaultOutputModes": ["text/plain", "application/json"],
    "skills": [
        {
            "id": "hello",
            "name": "Say Hello",
            "description": "Responds with a friendly 'Hello World' message",
            "tags": ["greeting", "hello", "simple"],
            "inputModes": ["text/plain", "application/json"],
            "outputModes": ["text/plain", "application/json"],
            "examples": ["Say hello", "Greet me", "Hello"]
        }
    ],
    "provider": {
        "organization": "A2A Registry Team",
        "url": "https://github.com/prassanna-ravishankar/a2a-registry"
    },
    "documentationUrl": "https://github.com/prassanna-ravishankar/a2a-registry",
    "author": "A2A Registry Team",
    "homepage": "https://hello.a2aregistry.org",
    "license": "MIT"
}

@app.get("/.well-known/agent-card.json")
async def agent_card(request: Request):
    host = request.headers.get("x-forwarded-host") or request.headers.get("host") or "localhost:7860"
    proto = request.headers.get("x-forwarded-proto") or "https"
    url_base = f"{proto}://{host}"
    card = dict(AGENT_CARD)
    card["url"] = AGENT_CARD["url"].replace("{HOST}", host).replace("https://{HOST}", url_base)
    return JSONResponse(card)

@app.get("/")
async def root():
    return {"status": "ok", "send_endpoint": "/a2a"}

@app.post("/a2a")
async def a2a_endpoint(request: Request):
    ctype = request.headers.get("content-type", "")
    if "application/json" in ctype:
        body = await request.json()
        msg = body.get("message") or body.get("text") or str(body)
        return JSONResponse(hello_agent(msg))
    text = (await request.body()).decode("utf-8", errors="ignore")
    return JSONResponse(hello_agent(text))