Spaces:
Sleeping
Sleeping
File size: 2,432 Bytes
126d4ef |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 |
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))
|