diff --git a/src/praisonai/praisonai/api/agent_invoke.py b/src/praisonai/praisonai/api/agent_invoke.py index d501a16edc..01b1e8e449 100644 --- a/src/praisonai/praisonai/api/agent_invoke.py +++ b/src/praisonai/praisonai/api/agent_invoke.py @@ -636,18 +636,18 @@ async def list_agents( @router.post("/agents/{agent_id}/register") async def register_agent_endpoint(agent_id: str, _: None = Depends(verify_token)) -> Dict[str, Any]: """ - Register an agent for API access. - - Note: This is a placeholder endpoint. In practice, agents would be - registered programmatically when they are created. + HTTP registration is not supported; agents are registered in-process. + + Returns 501 so clients can detect that nothing was registered. """ - # In a real implementation, this might load an agent from a configuration - # or database. For now, return information about registration. - return { - "message": f"Agent registration endpoint for '{agent_id}'", - "note": "Agents are typically registered programmatically", - "status": "info" - } + raise HTTPException( + status_code=501, + detail=( + f"Dynamic HTTP registration of '{agent_id}' is not supported; " + "register agents in-process via " + "praisonai.api.agent_invoke.register_agent(...)" + ), + ) @router.delete("/agents/{agent_id}") async def unregister_agent_endpoint( diff --git a/src/praisonai/praisonai/cli/features/serve.py b/src/praisonai/praisonai/cli/features/serve.py index f2dae78bc3..c01a79b4f4 100644 --- a/src/praisonai/praisonai/cli/features/serve.py +++ b/src/praisonai/praisonai/cli/features/serve.py @@ -645,11 +645,17 @@ def cmd_a2a(self, args: List[str]) -> int: "host": {"default": self.DEFAULT_HOST}, "port": {"default": 8082, "type": "int"}, "file": {"default": "agents.yaml"}, + "api_key": {"default": None}, } parsed = self._parse_args(args, spec) - + + if not os.path.exists(parsed["file"]): + self._print_error(f"Agents file not found: {parsed['file']}") + return self.EXIT_VALIDATION_ERROR + try: self._print_success(f"Starting A2A server on {parsed['host']}:{parsed['port']}") + print(f" Agents file: {parsed['file']}") print(" Agent Card: /.well-known/agent.json") print(" Discovery: /__praisonai__/discovery") @@ -664,18 +670,100 @@ def cmd_a2a(self, args: List[str]) -> int: return self.EXIT_GENERAL_ERROR return self.EXIT_SUCCESS - + + @staticmethod + def _yaml_query_runner(agents_file: str) -> Any: + """Return ``async run(query) -> str`` executing the agents YAML workflow.""" + import yaml + with open(agents_file) as f: + yaml_framework = (yaml.safe_load(f) or {}).get("framework") + + async def run(query: str) -> str: + import praisonai + result = await praisonai.arun( + agent_file=agents_file, + framework=yaml_framework, + cli_config={"topic": query} if query else None, + ) + return str(result) + + return run + + @staticmethod + def _add_a2a_route(app: Any, run_query: Any) -> None: + """Mount ``POST /a2a`` (JSON-RPC ``message/send``) backed by ``run_query``.""" + @app.post("/a2a") + async def a2a_endpoint(request_data: dict): + """Handle A2A JSON-RPC requests.""" + method = request_data.get("method") + params = request_data.get("params", {}) + request_id = request_data.get("id") + + if method != "message/send": + return { + "jsonrpc": "2.0", + "id": request_id, + "error": {"code": -32601, "message": f"Method not found: {method}"}, + } + + message = params.get("message") if isinstance(params, dict) else None + parts = message.get("parts") if isinstance(message, dict) else None + text = "".join( + p["text"] for p in (parts if isinstance(parts, list) else []) + if isinstance(p, dict) and p.get("type") == "text" and isinstance(p.get("text"), str) + ) + if not text: + return { + "jsonrpc": "2.0", + "id": request_id, + "error": {"code": -32602, "message": "message/send requires a text part"}, + } + try: + reply = await run_query(text) + except Exception as e: + return { + "jsonrpc": "2.0", + "id": request_id, + "error": {"code": -32603, "message": f"Agent execution failed: {e}"}, + } + return { + "jsonrpc": "2.0", + "id": request_id, + "result": { + "message": { + "role": "assistant", + "parts": [{"type": "text", "text": reply}], + }, + }, + } + + @staticmethod + def _resolve_api_key(config: Dict[str, Any], command: str) -> Optional[str]: + """Require an API key for non-localhost binds of agent-driving servers.""" + api_key = config.get("api_key") or os.environ.get("PRAISONAI_SERVE_API_KEY") + if config.get("host", ServeHandler.DEFAULT_HOST) not in _LOCALHOST_HOSTS and not api_key: + raise SystemExit( + f"praisonai serve {command}: --api-key (or PRAISONAI_SERVE_API_KEY) is " + "required when binding to a non-localhost host; it exposes " + "tool-driving routes." + ) + return api_key + def _create_a2a_app(self, config: Dict[str, Any]) -> Any: """Create FastAPI app for A2A server.""" from fastapi import FastAPI - + from praisonai.endpoints.discovery import ( create_discovery_document, EndpointInfo, ProviderInfo, ) from praisonai.endpoints.server import add_discovery_routes - + + # POST /a2a runs the YAML workflow (and its tools); gate non-localhost. + api_key = self._resolve_api_key(config, "a2a") + run_query = self._yaml_query_runner(config["file"]) + discovery = create_discovery_document(server_name="praisonai-a2a") discovery.add_provider(ProviderInfo( type="a2a", @@ -711,41 +799,9 @@ async def agent_card(): async def status(): """Return server status.""" return {"status": "ok", "name": "PraisonAI A2A", "version": "1.0.0"} - - @app.post("/a2a") - async def a2a_endpoint(request_data: dict): - """Handle A2A JSON-RPC requests.""" - method = request_data.get("method") - params = request_data.get("params", {}) - request_id = request_data.get("id") - - if method == "message/send": - # Handle message - message = params.get("message", {}) - parts = message.get("parts", []) - text = "" - for part in parts: - if part.get("type") == "text": - text += part.get("text", "") - - # TODO: Process with agent - return { - "jsonrpc": "2.0", - "id": request_id, - "result": { - "message": { - "role": "assistant", - "parts": [{"type": "text", "text": f"Received: {text}"}], - }, - }, - } - - return { - "jsonrpc": "2.0", - "id": request_id, - "error": {"code": -32601, "message": f"Method not found: {method}"}, - } - + + self._add_a2a_route(app, run_query) + discovery.add_endpoint(EndpointInfo( name="a2a", description="A2A protocol endpoint", @@ -760,7 +816,11 @@ async def root(): "agent_card": "/.well-known/agent.json", "discovery": "/__praisonai__/discovery", } - + + _install_api_key_middleware( + app, api_key, {"/", "/status", "/.well-known/agent.json"} + ) + return app def cmd_a2u(self, args: List[str]) -> int: @@ -892,19 +952,11 @@ def cmd_unified(self, args: List[str]) -> int: # Security: the unified server mounts tool-driving routes; refuse to bind # a non-localhost host without an API key, mirroring cmd_agents. - host = parsed.get("host", self.DEFAULT_HOST) - api_key = parsed.get("api_key") or os.environ.get("PRAISONAI_SERVE_API_KEY") - if host not in _LOCALHOST_HOSTS and not api_key: - raise SystemExit( - "praisonai serve unified: --api-key (or PRAISONAI_SERVE_API_KEY) is " - "required when binding to a non-localhost host; the unified server " - "exposes tool-driving routes." - ) - parsed["api_key"] = api_key + parsed["api_key"] = self._resolve_api_key(parsed, "unified") try: self._print_success(f"Starting unified server on {parsed['host']}:{parsed['port']}") - print(" Providers: agents-api, recipe, mcp, a2a, a2u") + print(f" Providers: {', '.join(self._unified_providers(parsed))}") print(" Discovery: /__praisonai__/discovery") # Lazy builder: built once (here when not reloading, or in the reload @@ -927,11 +979,22 @@ def cmd_unified(self, args: List[str]) -> int: return self.EXIT_GENERAL_ERROR return self.EXIT_SUCCESS - + + @staticmethod + def _unified_providers(config: Dict[str, Any]) -> List[str]: + """Providers the unified app actually mounts. + + agents-api and a2a need the agents YAML (--file); recipe and MCP run as + their own servers (``serve recipe`` / ``praisonai mcp serve``). + """ + if os.path.exists(config.get("file") or ""): + return ["agents-api", "a2a", "a2u"] + return ["a2u"] + def _create_unified_app(self, config: Dict[str, Any]) -> Any: - """Create unified FastAPI app with all providers.""" - from fastapi import FastAPI - + """Create unified FastAPI app with every provider it can mount.""" + from fastapi import FastAPI, HTTPException, Request + from praisonai.endpoints.discovery import ( create_discovery_document, EndpointInfo, @@ -939,49 +1002,60 @@ def _create_unified_app(self, config: Dict[str, Any]) -> Any: ) from praisonai.endpoints.server import add_discovery_routes from praisonai.endpoints.a2u_server import create_a2u_routes - + + providers = self._unified_providers(config) discovery = create_discovery_document(server_name="praisonai-unified") - - # Add all providers - discovery.add_provider(ProviderInfo( - type="agents-api", - name="Agents API", - capabilities=["invoke", "health"], - )) - discovery.add_provider(ProviderInfo( - type="recipe", - name="Recipe Runner", - capabilities=["list", "describe", "invoke", "stream"], - )) - discovery.add_provider(ProviderInfo( - type="mcp", - name="MCP Server", - capabilities=["list-tools", "call-tool"], - )) - discovery.add_provider(ProviderInfo( - type="a2a", - name="A2A Protocol", - capabilities=["agent-card", "message-send"], - )) - discovery.add_provider(ProviderInfo( - type="a2u", - name="A2U Event Stream", - capabilities=["subscribe", "stream"], - )) - app = FastAPI( title="PraisonAI Unified Server", description="Unified server with all PraisonAI providers", ) - add_discovery_routes(app, discovery) + + if "agents-api" in providers: + run_query = self._yaml_query_runner(config["file"]) + + @app.post("/agents") + async def invoke_agents(request: Request): + """Invoke the agents YAML workflow with a query.""" + try: + body = await request.json() + except Exception: + raise HTTPException(status_code=400, detail="Invalid request") + if not isinstance(body, dict): + raise HTTPException(status_code=400, detail="Request body must be a JSON object") + query = body.get("query", "") or body.get("message", "") + try: + return {"response": await run_query(query)} + except Exception as e: + raise HTTPException(status_code=500, detail=f"Agent execution failed: {e}") + + self._add_a2a_route(app, run_query) + discovery.add_provider(ProviderInfo( + type="agents-api", + name="Agents API", + capabilities=["invoke"], + )) + discovery.add_provider(ProviderInfo( + type="a2a", + name="A2A Protocol", + capabilities=["agent-card", "message-send"], + )) + discovery.add_endpoint(EndpointInfo( + name="agents", + description="Agents workflow endpoint", + provider_type="agents-api", + )) + discovery.add_endpoint(EndpointInfo( + name="a2a", + description="A2A protocol endpoint", + provider_type="a2a", + )) + create_a2u_routes(app) - - # Add sample endpoints - discovery.add_endpoint(EndpointInfo( - name="agents", - description="Agents workflow endpoint", - provider_type="agents-api", + discovery.add_provider(ProviderInfo( + type="a2u", + name="A2U Event Stream", + capabilities=["subscribe", "stream"], )) discovery.add_endpoint(EndpointInfo( name="events", @@ -994,17 +1068,23 @@ def _create_unified_app(self, config: Dict[str, Any]) -> Any: async def root(): return { "message": "PraisonAI Unified Server", - "providers": ["agents-api", "recipe", "mcp", "a2a", "a2u"], + "providers": providers, "discovery": "/__praisonai__/discovery", } - + @app.get("/.well-known/agent.json") async def agent_card(): - return { + card = { "name": "PraisonAI Unified", "description": "PraisonAI Unified Server", "version": "1.0.0", } + if "a2a" in providers: + card["url"] = ( + f"http://{config.get('host', 'localhost')}:" + f"{config.get('port', self.DEFAULT_PORT)}/a2a" + ) + return card _install_api_key_middleware(app, config.get("api_key")) diff --git a/src/praisonai/praisonai/endpoints/server.py b/src/praisonai/praisonai/endpoints/server.py index 2ec6231925..c46fa61f48 100644 --- a/src/praisonai/praisonai/endpoints/server.py +++ b/src/praisonai/praisonai/endpoints/server.py @@ -30,20 +30,22 @@ def add_discovery_routes( path: Path for discovery endpoint """ try: + from starlette.requests import Request from starlette.responses import JSONResponse from starlette.routing import Route except ImportError: # Try FastAPI try: + from fastapi import Request from fastapi.responses import JSONResponse except ImportError: raise ImportError("Starlette or FastAPI required for discovery routes") - async def discovery_handler(request): + async def discovery_handler(request: Request): """Return discovery document.""" return JSONResponse(discovery.to_dict()) - async def health_handler(request): + async def health_handler(request: Request): """Return health status with discovery info.""" return JSONResponse({ "status": "healthy", diff --git a/src/praisonai/tests/unit/test_serve_surfaces_honest.py b/src/praisonai/tests/unit/test_serve_surfaces_honest.py new file mode 100644 index 0000000000..cc7e0a9135 --- /dev/null +++ b/src/praisonai/tests/unit/test_serve_surfaces_honest.py @@ -0,0 +1,132 @@ +"""Regression tests: serve surfaces deliver what they advertise (issue #5973).""" + +import pytest + +pytest.importorskip("fastapi") +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from praisonai.cli.features.serve import ServeHandler + + +@pytest.fixture +def agents_file(tmp_path): + path = tmp_path / "agents.yaml" + path.write_text("framework: praisonai\nroles: {}\n") + return str(path) + + +@pytest.fixture +def fake_arun(monkeypatch): + import praisonai + + calls = [] + + async def _arun(**kwargs): + calls.append(kwargs) + return f"agent says: {kwargs['cli_config']['topic']}" + + monkeypatch.setattr(praisonai, "arun", _arun, raising=False) + return calls + + +def _send(client, text): + return client.post("/a2a", json={ + "jsonrpc": "2.0", "id": 1, "method": "message/send", + "params": {"message": {"parts": [{"type": "text", "text": text}]}}, + }).json() + + +def test_a2a_message_send_runs_agent(agents_file, fake_arun): + app = ServeHandler()._create_a2a_app( + {"host": "127.0.0.1", "port": 8082, "file": agents_file} + ) + body = _send(TestClient(app), "summarize this") + assert body["result"]["message"]["parts"][0]["text"] == "agent says: summarize this" + assert fake_arun[0]["agent_file"] == agents_file + assert fake_arun[0]["framework"] == "praisonai" + + +def test_a2a_missing_text_returns_invalid_params(agents_file, fake_arun): + client = TestClient(ServeHandler()._create_a2a_app( + {"host": "127.0.0.1", "port": 8082, "file": agents_file} + )) + for params in ({"message": {"parts": []}}, None, {"message": "x"}): + body = client.post("/a2a", json={ + "jsonrpc": "2.0", "id": 7, "method": "message/send", "params": params, + }).json() + assert body["id"] == 7 + assert body["error"]["code"] == -32602 + assert fake_arun == [] + + +def test_a2a_agent_failure_returns_internal_error(agents_file, monkeypatch): + import praisonai + + async def _boom(**kwargs): + raise RuntimeError("boom") + + monkeypatch.setattr(praisonai, "arun", _boom, raising=False) + body = _send(TestClient(ServeHandler()._create_a2a_app( + {"host": "127.0.0.1", "port": 8082, "file": agents_file} + )), "hi") + assert body["id"] == 1 + assert body["error"]["code"] == -32603 + assert "boom" in body["error"]["message"] + + +def test_a2a_requires_api_key_off_localhost(agents_file, monkeypatch): + monkeypatch.delenv("PRAISONAI_SERVE_API_KEY", raising=False) + with pytest.raises(SystemExit): + ServeHandler()._create_a2a_app( + {"host": "0.0.0.0", "port": 8082, "file": agents_file} + ) + + +def test_a2a_missing_file_rejected(tmp_path): + rc = ServeHandler().cmd_a2a(["--file", str(tmp_path / "missing.yaml")]) + assert rc == ServeHandler.EXIT_VALIDATION_ERROR + + +def _advertised_providers(client): + doc = client.get("/__praisonai__/discovery").json() + return {p["type"] for p in doc["providers"]} + + +def test_unified_mounts_every_advertised_provider(agents_file, fake_arun): + app = ServeHandler()._create_unified_app( + {"host": "127.0.0.1", "port": 8765, "file": agents_file} + ) + client = TestClient(app) + assert _advertised_providers(client) == {"agents-api", "a2a", "a2u"} + assert client.post("/agents", json={"query": "hi"}).json() == {"response": "agent says: hi"} + assert _send(client, "hey")["result"]["message"]["parts"][0]["text"] == "agent says: hey" + for bad in ([], None, "q"): + assert client.post("/agents", json=bad).status_code == 400 + + +def test_unified_without_yaml_only_advertises_a2u(tmp_path): + app = ServeHandler()._create_unified_app( + {"host": "127.0.0.1", "port": 8765, "file": str(tmp_path / "missing.yaml")} + ) + client = TestClient(app) + assert _advertised_providers(client) == {"a2u"} + assert client.post("/agents", json={"query": "hi"}).status_code in (404, 405) + + +def test_discovery_route_serves_document(): + app = ServeHandler()._create_a2u_app({"host": "127.0.0.1", "port": 8083}) + resp = TestClient(app).get("/__praisonai__/discovery") + assert resp.status_code == 200 + + +def test_http_register_is_not_a_silent_success(monkeypatch): + from praisonai.api import agent_invoke + + monkeypatch.setenv("CALL_SERVER_TOKEN", "t") + app = FastAPI() + app.include_router(agent_invoke.router) + resp = TestClient(app).post( + "/api/v1/agents/foo/register", headers={"Authorization": "Bearer t"} + ) + assert resp.status_code == 501