diff --git a/README.md b/README.md index fe8b5f8..b1edd56 100644 --- a/README.md +++ b/README.md @@ -92,6 +92,8 @@ ecp conformance --target "python examples/customer_support_demo/agent.py" ecp doctor ``` +Python agents may implement `@on_step` and `@on_reset` with regular functions or `async def`. See [`examples/async_python_demo`](examples/async_python_demo) for stdio and Streamable HTTP usage. For long-running calls, pass `--timeout` to `ecp run` or `ecp conformance`. + ## Repo Layout - `sdk/` - Python SDK for implementing ECP agents @@ -107,4 +109,3 @@ ecp doctor - Docs site: https://evaluationcontextprotocol.io/ - Quickstart: https://evaluationcontextprotocol.io/quickstart/ - Specification: https://evaluationcontextprotocol.io/spec/ - diff --git a/docs/examples.md b/docs/examples.md index bcf3cd9..dbea5eb 100644 --- a/docs/examples.md +++ b/docs/examples.md @@ -142,6 +142,25 @@ scenarios: message: "transport is online" ``` +## Async Python Agent Example + +Agent file: `examples/async_python_demo/agent.py` + +This example uses `async def` for both `@on_step` and `@on_reset`. The SDK awaits both hooks on a persistent event loop, allowing async clients to be reused safely between requests. + +Run it over stdio: + +```bash +ecp run --manifest examples/async_python_demo/manifest.yaml --timeout 10 +``` + +Or start the same agent over Streamable HTTP: + +```bash +ECP_TRANSPORT=http ECP_HTTP_PORT=8765 python examples/async_python_demo/agent.py +ecp conformance --target http://127.0.0.1:8765/ecp --timeout 10 +``` + ## LangChain Example Agent file: `examples/langchain_demo/agent.py` diff --git a/docs/quickstart.md b/docs/quickstart.md index 6c9ff99..caca434 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -106,9 +106,16 @@ For protocol implementers: ecp conformance --target "python examples/customer_support_demo/agent.py" ``` +For long-running agent calls, set a timeout explicitly: + +```bash +ecp run --manifest manifest.yaml --timeout 60 +ecp conformance --target "python agent.py" --timeout 60 +``` + ## Notes - The current release line is `0.3.1`. - New agents should use `evaluation_context`; `private_thought` remains a deprecated compatibility alias. -- Use `ECP_RPC_TIMEOUT` to control step timeouts. The default is 30 seconds. - +- `--timeout` controls the RPC timeout for `run` and `conformance`. It overrides `ECP_RPC_TIMEOUT`; the default is 30 seconds. +- Python SDK `@on_step` and `@on_reset` hooks may be synchronous or `async def` functions. diff --git a/examples/async_python_demo/agent.py b/examples/async_python_demo/agent.py new file mode 100644 index 0000000..aaa2087 --- /dev/null +++ b/examples/async_python_demo/agent.py @@ -0,0 +1,47 @@ +import asyncio +import os +import sys +from pathlib import Path + +SDK_SRC = Path(__file__).resolve().parents[2] / "sdk" / "python" / "src" +if str(SDK_SRC) not in sys.path: + sys.path.insert(0, str(SDK_SRC)) + +from ecp import Result, agent, on_reset, on_step, serve, serve_http + + +@agent(name="AsyncResearchAgent") +class AsyncResearchAgent: + def __init__(self) -> None: + self.completed_requests = 0 + + @on_step + async def step(self, user_input: str) -> Result: + query = (user_input or "").strip() + await asyncio.sleep(0.01) + self.completed_requests += 1 + return Result( + public_output=f"Async research complete for: {query}", + evaluation_context=( + f"Awaited the research source and completed request #{self.completed_requests}." + ), + tool_calls=[{"name": "async_research", "arguments": {"query": query}}], + ) + + @on_reset + async def reset(self) -> None: + await asyncio.sleep(0) + self.completed_requests = 0 + + +if __name__ == "__main__": + instance = AsyncResearchAgent() + if os.environ.get("ECP_TRANSPORT", "stdio").lower() == "http": + serve_http( + instance, + host=os.environ.get("ECP_HTTP_HOST", "127.0.0.1"), + port=int(os.environ.get("ECP_HTTP_PORT", "8765")), + path=os.environ.get("ECP_HTTP_PATH", "/ecp"), + ) + else: + serve(instance) diff --git a/examples/async_python_demo/manifest.yaml b/examples/async_python_demo/manifest.yaml new file mode 100644 index 0000000..759e3d3 --- /dev/null +++ b/examples/async_python_demo/manifest.yaml @@ -0,0 +1,23 @@ +manifest_version: "v1" +name: "Async Python Agent Validation" +target: "python examples/async_python_demo/agent.py" + +scenarios: + - name: "Awaited Research" + steps: + - input: "ECP async lifecycle support" + graders: + - type: text_match + field: public_output + condition: contains + value: "Async research complete" + + - type: text_match + field: evaluation_context + condition: contains + value: "completed request #1" + + - type: tool_usage + tool_name: "async_research" + arguments: + query: "ECP async lifecycle support" diff --git a/runtime/python/src/ecp_runtime/cli.py b/runtime/python/src/ecp_runtime/cli.py index 03797e8..7a2f189 100644 --- a/runtime/python/src/ecp_runtime/cli.py +++ b/runtime/python/src/ecp_runtime/cli.py @@ -19,7 +19,7 @@ ) from .manifest import ECPManifest from .reporter import HTMLReporter -from .runner import ECPRunner +from .runner import ECPRunner, resolve_rpc_timeout from .trend import RunTrendAnalyzer app = typer.Typer( @@ -85,6 +85,11 @@ def run( "--fail-on-error/--no-fail-on-error", help="Exit non-zero if any checks fail (useful for CI)", ), + timeout: Optional[float] = typer.Option( + None, + "--timeout", + help="RPC timeout in seconds (overrides ECP_RPC_TIMEOUT)", + ), ): """ Execute an evaluation run based on a manifest file. @@ -99,7 +104,7 @@ def run( config = ECPManifest.from_yaml(str(manifest)) # Run the Tests - runner = ECPRunner(config) + runner = ECPRunner(config, rpc_timeout=timeout) result_summary = runner.run_scenarios() total = int(result_summary.get("total", 0) or 0) passed = int(result_summary.get("passed", 0) or 0) @@ -245,12 +250,21 @@ def conformance( "--json", help="Print only the machine-readable conformance report", ), + timeout: Optional[float] = typer.Option( + None, + "--timeout", + help="RPC timeout in seconds (overrides ECP_RPC_TIMEOUT)", + ), ): """ Validate the core ECP protocol contract against an agent. """ - runner = ECPRunner(type("Manifest", (), {"target": target, "scenarios": []})()) - agent = runner._create_agent(target, rpc_timeout=float(os.environ.get("ECP_RPC_TIMEOUT", "30"))) + rpc_timeout = resolve_rpc_timeout(timeout) + runner = ECPRunner( + type("Manifest", (), {"target": target, "scenarios": []})(), + rpc_timeout=rpc_timeout, + ) + agent = runner._create_agent(target, rpc_timeout=rpc_timeout) checks: List[Dict[str, Any]] = [] started = False try: diff --git a/runtime/python/src/ecp_runtime/runner.py b/runtime/python/src/ecp_runtime/runner.py index 68ec46f..071607e 100644 --- a/runtime/python/src/ecp_runtime/runner.py +++ b/runtime/python/src/ecp_runtime/runner.py @@ -2,11 +2,11 @@ Docstring for runtime.python.src.ecp_runtime.runner Simplified Version. V0.1 -AsyncIO Pending """ import json import logging +import math import os import queue import subprocess @@ -235,8 +235,9 @@ def _parse_sse_response(self, raw: str) -> Dict[str, Any]: class ECPRunner: """The Orchestrator.""" - def __init__(self, manifest): + def __init__(self, manifest, rpc_timeout: Optional[float] = None): self.manifest = manifest + self.rpc_timeout = resolve_rpc_timeout(rpc_timeout) def run_scenarios(self): total_passed = 0 @@ -246,13 +247,17 @@ def run_scenarios(self): for scenario in self.manifest.scenarios: logger.info("Scenario: %s", scenario.name) - rpc_timeout = float(os.environ.get("ECP_RPC_TIMEOUT", "30")) - agent = self._create_agent(self.manifest.target, rpc_timeout=rpc_timeout) + agent = self._create_agent(self.manifest.target, rpc_timeout=self.rpc_timeout) agent.start() try: - init_resp = agent.send_rpc("agent/initialize", {"config": {}}) - self._ensure_rpc_success(init_resp, scenario.name, step_idx=None, method="agent/initialize") + init_resp = self._call_rpc( + agent, + "agent/initialize", + {"config": {}}, + scenario.name, + step_idx=None, + ) try: validate_initialize_result(init_resp["result"]) except ValueError as exc: @@ -263,8 +268,13 @@ def run_scenarios(self): for i, step in enumerate(scenario.steps): # Execute - rpc_resp = agent.send_rpc("agent/step", {"input": step.input}) - self._ensure_rpc_success(rpc_resp, scenario.name, step_idx=i + 1, method="agent/step") + rpc_resp = self._call_rpc( + agent, + "agent/step", + {"input": step.input}, + scenario.name, + step_idx=i + 1, + ) result_data = validate_step_result(rpc_resp["result"]) # Map to internal object @@ -340,6 +350,38 @@ def _ensure_rpc_success( except ValueError as exc: raise RuntimeError(f"RPC call failed ({method}) at {where}: {exc}") from exc + def _call_rpc( + self, + agent: Any, + method: str, + params: Dict[str, Any], + scenario_name: str, + step_idx: Optional[int], + ) -> Dict[str, Any]: + where = f"scenario='{scenario_name}'" + if step_idx is not None: + where += f", step={step_idx}" + try: + response = agent.send_rpc(method, params) + except Exception as exc: + raise RuntimeError(f"RPC call failed ({method}) at {where}: {exc}") from exc + self._ensure_rpc_success(response, scenario_name, step_idx, method) + return response + + +def resolve_rpc_timeout(value: Optional[float] = None) -> float: + """Resolve and validate an explicit timeout or the ECP_RPC_TIMEOUT fallback.""" + raw_value: Any = value if value is not None else os.environ.get("ECP_RPC_TIMEOUT", "30") + try: + timeout = float(raw_value) + except (TypeError, ValueError) as exc: + raise ValueError( + f"RPC timeout must be a positive number; received {raw_value!r}" + ) from exc + if not math.isfinite(timeout) or timeout <= 0: + raise ValueError(f"RPC timeout must be a positive finite number; received {raw_value!r}") + return timeout + def _is_http_url(target: str) -> bool: parsed = urlparse(target) diff --git a/runtime/python/tests/test_cli.py b/runtime/python/tests/test_cli.py index af38b86..82683da 100644 --- a/runtime/python/tests/test_cli.py +++ b/runtime/python/tests/test_cli.py @@ -9,6 +9,7 @@ if str(RUNTIME_SRC) not in sys.path: sys.path.insert(0, str(RUNTIME_SRC)) +import ecp_runtime.cli as cli_module from ecp_runtime.cli import app from typer.testing import CliRunner @@ -72,6 +73,26 @@ def test_failure_exit_code(self) -> None: result = self.runner.invoke(app, ["run", "--manifest", self.manifest_path, "--no-fail-on-error"]) self.assertEqual(result.exit_code, 0, msg=result.output) + def test_run_passes_explicit_timeout_to_runner(self) -> None: + fake_config = object() + runtime_class = mock.Mock( + return_value=mock.Mock( + run_scenarios=mock.Mock(return_value={"passed": 0, "total": 0, "scenarios": []}) + ) + ) + with mock.patch("ecp_runtime.cli._configure_logging"), mock.patch.object( + cli_module.ECPManifest, + "from_yaml", + return_value=fake_config, + ), mock.patch("ecp_runtime.cli.ECPRunner", runtime_class): + result = self.runner.invoke( + app, + ["run", "--manifest", self.manifest_path, "--timeout", "4.5"], + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + runtime_class.assert_called_once_with(fake_config, rpc_timeout=4.5) + def test_validate_command(self) -> None: result = self.runner.invoke(app, ["validate", self.manifest_path]) @@ -114,6 +135,26 @@ def test_conformance_json_report(self) -> None: self.assertEqual(payload["total"], 3) fake_agent.stop.assert_called_once() + def test_conformance_passes_explicit_timeout_to_transport(self) -> None: + fake_agent = mock.Mock() + fake_agent.send_rpc.side_effect = [ + {"jsonrpc": "2.0", "id": 1, "result": {"name": "test", "capabilities": {}}}, + {"jsonrpc": "2.0", "id": 2, "result": {"status": "done"}}, + {"jsonrpc": "2.0", "id": 3, "result": True}, + ] + fake_runtime = mock.Mock() + fake_runtime._create_agent.return_value = fake_agent + runtime_class = mock.Mock(return_value=fake_runtime) + + with mock.patch("ecp_runtime.cli.ECPRunner", runtime_class): + result = self.runner.invoke( + app, + ["conformance", "--target", "python agent.py", "--timeout", "6", "--json"], + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + fake_runtime._create_agent.assert_called_once_with("python agent.py", rpc_timeout=6.0) + def test_conformance_rejects_invalid_step_result(self) -> None: fake_agent = mock.Mock() fake_agent.send_rpc.side_effect = [ diff --git a/runtime/python/tests/test_example_integrations.py b/runtime/python/tests/test_example_integrations.py index d75b3a7..a098416 100644 --- a/runtime/python/tests/test_example_integrations.py +++ b/runtime/python/tests/test_example_integrations.py @@ -50,6 +50,62 @@ def test_two_agent_demo_manifest_passes(self) -> None: self.assertEqual(result.returncode, 0, msg=result.stdout + result.stderr) self.assertIn('"failed": 0', result.stdout) + def test_async_python_demo_manifest_passes(self) -> None: + manifest = REPO_ROOT / "examples" / "async_python_demo" / "manifest.yaml" + result = self._run_manifest(manifest) + self.assertEqual(result.returncode, 0, msg=result.stdout + result.stderr) + self.assertIn('"failed": 0', result.stdout) + + def test_async_python_demo_conforms_over_http(self) -> None: + port = self._free_port() + env = dict(os.environ) + existing = env.get("PYTHONPATH", "") + extra_paths = [str(RUNTIME_SRC), str(REPO_ROOT / "sdk" / "python" / "src")] + if existing: + extra_paths.append(existing) + env["PYTHONPATH"] = os.pathsep.join(extra_paths) + env["ECP_TRANSPORT"] = "http" + env["ECP_HTTP_PORT"] = str(port) + + server = subprocess.Popen( + [sys.executable, str(REPO_ROOT / "examples" / "async_python_demo" / "agent.py")], + cwd=REPO_ROOT, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=env, + ) + + try: + self._wait_for_port("127.0.0.1", port) + result = subprocess.run( + [ + sys.executable, + "-m", + "ecp_runtime.cli", + "conformance", + "--target", + f"http://127.0.0.1:{port}/ecp", + "--timeout", + "5", + "--json", + ], + cwd=REPO_ROOT, + text=True, + capture_output=True, + env=env, + check=False, + ) + self.assertEqual(result.returncode, 0, msg=result.stdout + result.stderr) + self.assertIn('"conformant": true', result.stdout) + finally: + server.terminate() + try: + server.communicate(timeout=2) + except subprocess.TimeoutExpired: + server.kill() + server.communicate() + def test_streamable_http_demo_manifest_passes(self) -> None: port = self._free_port() server_env = dict(os.environ) diff --git a/runtime/python/tests/test_runner.py b/runtime/python/tests/test_runner.py index 377b99b..a937540 100644 --- a/runtime/python/tests/test_runner.py +++ b/runtime/python/tests/test_runner.py @@ -10,7 +10,12 @@ sys.path.insert(0, str(RUNTIME_SRC)) from ecp_runtime.manifest import StepConfig -from ecp_runtime.runner import ECPRunner, HTTPAgentClient, _ensure_response_id +from ecp_runtime.runner import ( + ECPRunner, + HTTPAgentClient, + _ensure_response_id, + resolve_rpc_timeout, +) class RunnerTests(unittest.TestCase): @@ -101,6 +106,66 @@ def test_runner_uses_http_client_for_url_target(self) -> None: self.assertEqual(agent.endpoint, "http://127.0.0.1:8765/ecp") self.assertEqual(agent.rpc_timeout, 12.0) + def test_runner_passes_explicit_timeout_to_agent(self) -> None: + observed = {} + + class FakeAgentProcess: + def __init__(self, command, rpc_timeout=30.0): + observed["timeout"] = rpc_timeout + + def start(self): + return None + + def stop(self): + return None + + def send_rpc(self, method, params=None): + if method == "agent/initialize": + return {"jsonrpc": "2.0", "id": 1, "result": {"name": "x", "capabilities": {}}} + return {"jsonrpc": "2.0", "id": 2, "result": {"status": "done"}} + + with mock.patch("ecp_runtime.runner.AgentProcess", FakeAgentProcess): + ECPRunner(self._manifest(), rpc_timeout=4.25).run_scenarios() + + self.assertEqual(observed["timeout"], 4.25) + + def test_transport_timeout_includes_method_and_step_context(self) -> None: + class FakeAgentProcess: + def __init__(self, command, rpc_timeout=30.0): + pass + + def start(self): + return None + + def stop(self): + return None + + def send_rpc(self, method, params=None): + if method == "agent/initialize": + return {"jsonrpc": "2.0", "id": 1, "result": {"name": "x", "capabilities": {}}} + raise RuntimeError("Agent response timed out after 0.1s") + + with mock.patch("ecp_runtime.runner.AgentProcess", FakeAgentProcess): + with self.assertRaises(RuntimeError) as ctx: + ECPRunner(self._manifest(), rpc_timeout=0.1).run_scenarios() + + message = str(ctx.exception) + self.assertIn("agent/step", message) + self.assertIn("Scenario A", message) + self.assertIn("step=1", message) + self.assertIn("timed out", message) + + def test_timeout_resolution_prefers_explicit_value_and_validates_environment(self) -> None: + with mock.patch.dict("os.environ", {"ECP_RPC_TIMEOUT": "8.5"}): + self.assertEqual(resolve_rpc_timeout(), 8.5) + self.assertEqual(resolve_rpc_timeout(2), 2.0) + + for invalid in ("invalid", "0", "-1", "inf", "nan"): + with self.subTest(invalid=invalid): + with mock.patch.dict("os.environ", {"ECP_RPC_TIMEOUT": invalid}): + with self.assertRaisesRegex(ValueError, "positive"): + resolve_rpc_timeout() + def test_http_agent_client_posts_json_rpc(self) -> None: class FakeResponse: headers = {"Content-Type": "application/json; charset=utf-8"} diff --git a/sdk/python/README.md b/sdk/python/README.md index 70907fe..cfbe58a 100644 --- a/sdk/python/README.md +++ b/sdk/python/README.md @@ -42,6 +42,34 @@ if __name__ == "__main__": `evaluation_context` is the preferred field for evaluator-safe audit evidence. `private_thought` is still accepted as a deprecated compatibility alias. +## Async lifecycle hooks + +`@on_step` and `@on_reset` may be synchronous functions or `async def` coroutines. The SDK awaits async hooks on one persistent event loop, so loop-bound clients and other async resources can be reused across requests. + +```python +import asyncio + +from ecp import Result, agent, on_reset, on_step, serve + + +@agent(name="AsyncAgent") +class AsyncAgent: + @on_step + async def step(self, user_input: str) -> Result: + await asyncio.sleep(0.01) + return Result(public_output=f"Processed: {user_input}") + + @on_reset + async def reset(self) -> None: + await asyncio.sleep(0) + + +if __name__ == "__main__": + serve(AsyncAgent()) +``` + +The same hooks work with `serve_http(...)`. See `examples/async_python_demo` for a runnable agent that supports either transport. + ## Streamable HTTP Agents can also run as an ECP Streamable HTTP server: @@ -58,4 +86,3 @@ The endpoint accepts JSON-RPC `POST` requests at `/ecp`. It returns JSON for req - Documentation: https://evaluationcontextprotocol.io/ - Repository: https://github.com/evaluation-context-protocol/ecp - Issues: https://github.com/evaluation-context-protocol/ecp/issues - diff --git a/sdk/python/src/ecp/server.py b/sdk/python/src/ecp/server.py index 9adf50e..d27c312 100644 --- a/sdk/python/src/ecp/server.py +++ b/sdk/python/src/ecp/server.py @@ -1,5 +1,9 @@ +import asyncio +import inspect import json import sys +import threading +from concurrent.futures import Future from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any, Dict, Iterable, List, Optional, Tuple @@ -11,6 +15,61 @@ JSON_CONTENT_TYPE = "application/json" +class _AwaitableExecutor: + """Runs async hooks on one persistent event loop without changing the sync API.""" + + def __init__(self) -> None: + self._loop: Optional[asyncio.AbstractEventLoop] = None + self._thread: Optional[threading.Thread] = None + self._lock = threading.Lock() + + def run(self, value: Any) -> Any: + if not inspect.isawaitable(value): + return value + loop = self._ensure_loop() + future: Future = asyncio.run_coroutine_threadsafe(self._await_value(value), loop) + return future.result() + + def close(self) -> None: + with self._lock: + loop = self._loop + thread = self._thread + self._loop = None + self._thread = None + if loop is None or thread is None: + return + loop.call_soon_threadsafe(loop.stop) + thread.join() + loop.close() + + def _ensure_loop(self) -> asyncio.AbstractEventLoop: + with self._lock: + if self._loop is not None and self._thread is not None and self._thread.is_alive(): + return self._loop + + ready = threading.Event() + loop = asyncio.new_event_loop() + + def _run_loop() -> None: + asyncio.set_event_loop(loop) + ready.set() + loop.run_forever() + + thread = threading.Thread(target=_run_loop, name="ecp-async-hooks", daemon=True) + thread.start() + ready.wait() + self._loop = loop + self._thread = thread + return loop + + @staticmethod + async def _await_value(value: Any) -> Any: + return await value + + +_ASYNC_EXECUTOR = _AwaitableExecutor() + + def serve(agent_instance): """ Starts the ECP Server loop. @@ -20,24 +79,27 @@ def serve(agent_instance): _CURRENT_AGENT_INSTANCE = agent_instance # 1. Input Loop (Reads 1 line at a time from the Runtime) - for line in sys.stdin: - if not line.strip(): - continue - req_id = None - - try: - request = json.loads(line) - if isinstance(request, dict): - req_id = request.get("id") - response = _dispatch_json_rpc(request) - if response is not None: - _write_json_rpc(response) + try: + for line in sys.stdin: + if not line.strip(): + continue + req_id = None - except Exception as e: - # If the agent crashes, we must tell the Runtime why - error_msg = f"{type(e).__name__}: {str(e)}" - # traceback.print_exc(file=sys.stderr) # Debugging help - _send_error(req_id if 'req_id' in locals() else None, -32000, error_msg) + try: + request = json.loads(line) + if isinstance(request, dict): + req_id = request.get("id") + response = _dispatch_json_rpc(request) + if response is not None: + _write_json_rpc(response) + + except Exception as e: + # If the agent crashes, we must tell the Runtime why + error_msg = f"{type(e).__name__}: {str(e)}" + # traceback.print_exc(file=sys.stderr) # Debugging help + _send_error(req_id if 'req_id' in locals() else None, -32000, error_msg) + finally: + _ASYNC_EXECUTOR.close() def serve_http( @@ -59,7 +121,11 @@ def serve_http( _CURRENT_AGENT_INSTANCE = agent_instance server = _build_http_server(host, port, path, allowed_origins) print(f"ECP Streamable HTTP listening on http://{host}:{server.server_port}{_normalize_path(path)}", file=sys.stderr) - server.serve_forever() + try: + server.serve_forever() + finally: + server.server_close() + _ASYNC_EXECUTOR.close() def _build_http_server( @@ -182,7 +248,7 @@ def _handle_step(params): user_input = params.get("input") # Execute User Logic - result = handler(user_input) + result = _invoke_handler(handler, user_input) # Ensure it returns a Result object if not isinstance(result, Result): @@ -202,10 +268,14 @@ def _handle_reset(): method_name = _HOOKS["reset"] if method_name: handler = getattr(_CURRENT_AGENT_INSTANCE, method_name) - handler() + _invoke_handler(handler) return True +def _invoke_handler(handler, *args): + return _ASYNC_EXECUTOR.run(handler(*args)) + + def _dispatch_http_payload(payload: Any) -> Optional[Any]: if isinstance(payload, list): responses: List[Dict[str, Any]] = [] diff --git a/sdk/python/tests/test_server.py b/sdk/python/tests/test_server.py index fdb71ce..7e4b308 100644 --- a/sdk/python/tests/test_server.py +++ b/sdk/python/tests/test_server.py @@ -1,3 +1,4 @@ +import asyncio import json import sys import threading @@ -11,7 +12,7 @@ sys.path.insert(0, str(SDK_SRC)) import ecp.server as server -from ecp import Result, agent, on_step +from ecp import Result, agent, on_reset, on_step @agent(name="HTTPTestAgent") @@ -21,6 +22,26 @@ def step(self, user_input: str) -> Result: return Result(public_output=f"echo: {user_input}", evaluation_context="echoed input") +@agent(name="AsyncHTTPTestAgent") +class AsyncHTTPTestAgent: + def __init__(self) -> None: + self.loop_ids = [] + self.reset_called = False + + @on_step + async def step(self, user_input: str) -> Result: + await asyncio.sleep(0) + self.loop_ids.append(id(asyncio.get_running_loop())) + if user_input == "fail": + raise RuntimeError("async failure") + return Result(public_output=f"async: {user_input}", evaluation_context="awaited input") + + @on_reset + async def reset(self) -> None: + await asyncio.sleep(0) + self.reset_called = True + + class StreamableHTTPServerTests(unittest.TestCase): def setUp(self) -> None: server._CURRENT_AGENT_INSTANCE = HTTPTestAgent() @@ -38,6 +59,7 @@ def tearDown(self) -> None: self.httpd.shutdown() self.thread.join(timeout=2) self.httpd.server_close() + server._ASYNC_EXECUTOR.close() def _post(self, payload, headers=None): req = request.Request( @@ -122,6 +144,46 @@ def test_rejects_untrusted_origin(self) -> None: self.assertEqual(ctx.exception.code, 403) + def test_async_step_uses_persistent_event_loop_over_http(self) -> None: + async_agent = AsyncHTTPTestAgent() + server._CURRENT_AGENT_INSTANCE = async_agent + + first_status, first_body, _headers = self._post( + {"jsonrpc": "2.0", "id": 1, "method": "agent/step", "params": {"input": "one"}} + ) + second_status, second_body, _headers = self._post( + {"jsonrpc": "2.0", "id": 2, "method": "agent/step", "params": {"input": "two"}} + ) + + self.assertEqual(first_status, 200) + self.assertEqual(second_status, 200) + self.assertEqual(first_body["result"]["public_output"], "async: one") + self.assertEqual(second_body["result"]["public_output"], "async: two") + self.assertEqual(len(set(async_agent.loop_ids)), 1) + + def test_async_reset_is_awaited(self) -> None: + async_agent = AsyncHTTPTestAgent() + server._CURRENT_AGENT_INSTANCE = async_agent + + status, body, _headers = self._post( + {"jsonrpc": "2.0", "id": 3, "method": "agent/reset", "params": {}} + ) + + self.assertEqual(status, 200) + self.assertTrue(body["result"]) + self.assertTrue(async_agent.reset_called) + + def test_async_exception_returns_json_rpc_error(self) -> None: + server._CURRENT_AGENT_INSTANCE = AsyncHTTPTestAgent() + + status, body, _headers = self._post( + {"jsonrpc": "2.0", "id": 4, "method": "agent/step", "params": {"input": "fail"}} + ) + + self.assertEqual(status, 200) + self.assertEqual(body["error"]["code"], -32000) + self.assertIn("async failure", body["error"]["message"]) + if __name__ == "__main__": unittest.main()