A2AServer(agent_card: AgentCard, *, handler: Optional[Callable[[str], str]] = None, bus: Optional[EventBus] = None)
A2A server that processes incoming tasks via agent execution.
Can be mounted as routes in the FastAPI server.
Source code in src/openjarvis/a2a/server.py
| def __init__(
self,
agent_card: AgentCard,
*,
handler: Optional[Callable[[str], str]] = None,
bus: Optional[EventBus] = None,
) -> None:
self._card = agent_card
self._handler = handler
self._bus = bus
self._tasks: Dict[str, A2ATask] = {}
|
Functions
handle_request(request_data: Dict[str, Any]) -> Dict[str, Any]
Process a JSON-RPC 2.0 A2A request.
Source code in src/openjarvis/a2a/server.py
| def handle_request(self, request_data: Dict[str, Any]) -> Dict[str, Any]:
"""Process a JSON-RPC 2.0 A2A request."""
method = request_data.get("method", "")
params = request_data.get("params", {})
req_id = request_data.get("id", "")
if method == "tasks/send":
return self._handle_task_send(params, req_id)
elif method == "tasks/get":
return self._handle_task_get(params, req_id)
elif method == "tasks/cancel":
return self._handle_task_cancel(params, req_id)
else:
return A2AResponse(
error={"code": -32601, "message": f"Method not found: {method}"},
request_id=req_id,
).to_dict()
|
get_routes() -> List[Dict[str, Any]]
Return route definitions for mounting in a web framework.
Source code in src/openjarvis/a2a/server.py
| def get_routes(self) -> List[Dict[str, Any]]:
"""Return route definitions for mounting in a web framework."""
return [
{
"path": "/.well-known/agent.json",
"method": "GET",
"handler": "agent_card",
},
{"path": "/a2a/tasks", "method": "POST", "handler": "handle_request"},
]
|