minor: add conformance and async lifecycle support

.github/workflows/tests.yml

@@ -9,7 +9,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.10", "3.11", "3.12"]
+        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
 
     steps:
       - uses: actions/checkout@v4

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/ci.md

@@ -49,3 +49,11 @@ ecp validate evals/support.yaml
 ecp conformance --target "python agent.py"
 ```
 
+For CI systems, emit a stable machine-readable report:
+
+```bash
+ecp conformance --target "python agent.py" --json-out conformance.json
+```
+
+The command exits with code `1` when any initialize, step-result, or reset
+contract check fails. Use `--json` to print only the JSON report to stdout.

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.

docs/spec.md

@@ -31,6 +31,7 @@ For Streamable HTTP, the agent runs as an HTTP server and exposes one endpoint,
 - `evaluation_context`: string or null
 - `private_thought`: deprecated compatibility alias for `evaluation_context`
 - `tool_calls`: array or null
+- `logs`: optional evaluator-visible execution logs or null
 
 Tool call format:
 

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

@@ -5,14 +5,21 @@
 import shutil
 import sys
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional
 
 import typer
 from pydantic import ValidationError
 
+from .conformance import (
+    build_conformance_report,
+    conformance_check,
+    validate_initialize_result,
+    validate_reset_result,
+    validate_step_result,
+)
 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(
@@ -78,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.
@@ -92,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)
@@ -227,32 +239,96 @@ def conformance(
         "--input",
         help="Input text for the conformance step call",
     ),
+    json_out: Optional[Path] = typer.Option(
+        None,
+        "--json-out",
+        help="Path to save the machine-readable conformance report",
+        resolve_path=True,
+    ),
+    print_json: bool = typer.Option(
+        False,
+        "--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)",
+    ),
 ):
     """
-    Run a small protocol conformance smoke test against an ECP agent.
+    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")))
-    agent.start()
+    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:
-        init_resp = agent.send_rpc("agent/initialize", {"config": {}})
-        _assert_rpc_result(init_resp, "agent/initialize")
-        step_resp = agent.send_rpc("agent/step", {"input": step_input})
-        _assert_rpc_result(step_resp, "agent/step")
-        result = step_resp.get("result", {})
-        if not isinstance(result, dict):
-            raise ValueError("agent/step result must be an object")
-        if "public_output" not in result and "evaluation_context" not in result and "tool_calls" not in result:
-            raise ValueError("agent/step result should expose public_output, evaluation_context, or tool_calls")
-        reset_resp = agent.send_rpc("agent/reset", {})
-        _assert_rpc_result(reset_resp, "agent/reset")
+        agent.start()
+        started = True
+        initialize = _run_conformance_call(
+            agent,
+            "initialize response",
+            "agent/initialize",
+            {"config": {}},
+            result_validator=validate_initialize_result,
+        )
+        checks.append(initialize)
+        if initialize["passed"]:
+            checks.append(
+                _run_conformance_call(
+                    agent,
+                    "step result contract",
+                    "agent/step",
+                    {"input": step_input},
+                    result_validator=validate_step_result,
+                )
+            )
+        else:
+            checks.append(_skipped_conformance_check("step result contract", "agent/step"))
     except Exception as exc:
-        typer.echo(f"Conformance failed: {exc}", err=True)
-        raise typer.Exit(code=1)
+        checks.append(
+            {
+                "name": "initialize response",
+                "method": "agent/initialize",
+                "passed": False,
+                "message": f"agent could not start: {exc}",
+            }
+        )
+        checks.append(_skipped_conformance_check("step result contract", "agent/step"))
     finally:
-        agent.stop()
-
-    typer.echo("Conformance smoke test passed")
+        if started:
+            checks.append(
+                _run_conformance_call(
+                    agent,
+                    "reset response",
+                    "agent/reset",
+                    {},
+                    result_validator=validate_reset_result,
+                )
+            )
+            agent.stop()
+        else:
+            checks.append(_skipped_conformance_check("reset response", "agent/reset"))
+
+    report = build_conformance_report(target, checks)
+    rendered = json.dumps(report, indent=2)
+    if json_out:
+        json_out.write_text(rendered + "\n", encoding="utf-8")
+    if print_json:
+        typer.echo(rendered)
+    else:
+        for check in checks:
+            marker = "PASS" if check["passed"] else "FAIL"
+            typer.echo(f"{marker} | {check['name']} | {check['message']}")
+        typer.echo(f"Conformance: {report['passed']}/{report['total']} checks passed")
+
+    if not report["conformant"]:
+        raise typer.Exit(code=1)
 
 
 @app.command()
@@ -316,15 +392,28 @@ def trend(
         raise typer.Exit(code=2)
 
 
-def _assert_rpc_result(response: Dict[str, Any], method: str) -> None:
-    if not isinstance(response, dict):
-        raise ValueError(f"{method} response must be a JSON-RPC object")
-    if response.get("jsonrpc") != "2.0":
-        raise ValueError(f"{method} response must include jsonrpc='2.0'")
-    if "error" in response:
-        raise ValueError(f"{method} returned error: {response['error']}")
-    if "result" not in response:
-        raise ValueError(f"{method} response must include result")
+def _run_conformance_call(
+    agent: Any,
+    name: str,
+    method: str,
+    params: Dict[str, Any],
+    *,
+    result_validator: Optional[Callable[[Any], Any]] = None,
+) -> Dict[str, Any]:
+    try:
+        response = agent.send_rpc(method, params)
+    except Exception as exc:
+        return {"name": name, "method": method, "passed": False, "message": str(exc)}
+    return conformance_check(name, method, response, result_validator=result_validator)
+
+
+def _skipped_conformance_check(name: str, method: str) -> Dict[str, Any]:
+    return {
+        "name": name,
+        "method": method,
+        "passed": False,
+        "message": "skipped because a prerequisite failed",
+    }
 
 
 def _starter_agent() -> str: