minor: add async agent lifecycle support

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/
-

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`

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.

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)

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"

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:

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)

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 = [