diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 7a5c391..239978d 100644 --- a/.github/workflows/tests.yml +++ b/.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 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/ci.md b/docs/ci.md index e902c7d..05cf126 100644 --- a/docs/ci.md +++ b/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. 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/docs/spec.md b/docs/spec.md index e5e9922..409145d 100644 --- a/docs/spec.md +++ b/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: 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 7c6ef38..7a2f189 100644 --- a/runtime/python/src/ecp_runtime/cli.py +++ b/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: diff --git a/runtime/python/src/ecp_runtime/conformance.py b/runtime/python/src/ecp_runtime/conformance.py new file mode 100644 index 0000000..f8d4831 --- /dev/null +++ b/runtime/python/src/ecp_runtime/conformance.py @@ -0,0 +1,108 @@ +"""Protocol contract validation shared by runtime execution and conformance checks.""" + +from typing import Any, Callable, Dict, List, Optional + +VALID_STATUSES = {"done", "paused"} + + +def validate_rpc_response(response: Any, method: str) -> Any: + """Validate the JSON-RPC response envelope and return its result.""" + 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 "id" not in response: + raise ValueError(f"{method} response must include id") + if "result" in response and "error" in response: + raise ValueError(f"{method} response cannot include both result and error") + if "error" in response: + error = response["error"] + if not isinstance(error, dict): + raise ValueError(f"{method} error must be an object") + if not isinstance(error.get("code"), int) or not isinstance(error.get("message"), str): + raise ValueError(f"{method} error must include integer code and string message") + raise ValueError(f"{method} returned error: {error}") + if "result" not in response: + raise ValueError(f"{method} response must include result") + return response["result"] + + +def validate_step_result(result: Any) -> Dict[str, Any]: + """Validate the normative agent/step result contract.""" + if not isinstance(result, dict): + raise ValueError("agent/step result must be an object") + + status = result.get("status") + if status not in VALID_STATUSES: + allowed = ", ".join(sorted(VALID_STATUSES)) + raise ValueError(f"agent/step result status must be one of: {allowed}") + + for field in ("public_output", "evaluation_context", "private_thought", "logs"): + value = result.get(field) + if value is not None and not isinstance(value, str): + raise ValueError(f"agent/step result {field} must be a string or null") + + tool_calls = result.get("tool_calls") + if tool_calls is not None: + if not isinstance(tool_calls, list): + raise ValueError("agent/step result tool_calls must be an array or null") + for index, tool_call in enumerate(tool_calls): + _validate_tool_call(tool_call, index) + + return result + + +def validate_initialize_result(result: Any) -> Dict[str, Any]: + if not isinstance(result, dict): + raise ValueError("agent/initialize result must be an object") + if not isinstance(result.get("name"), str) or not result["name"]: + raise ValueError("agent/initialize result name must be a non-empty string") + if not isinstance(result.get("capabilities"), dict): + raise ValueError("agent/initialize result capabilities must be an object") + return result + + +def validate_reset_result(result: Any) -> bool: + if result is not True: + raise ValueError("agent/reset result must be true") + return result + + +def conformance_check( + name: str, + method: str, + response: Any, + *, + result_validator: Optional[Callable[[Any], Any]] = None, +) -> Dict[str, Any]: + """Return a stable, serializable result for one protocol check.""" + try: + result = validate_rpc_response(response, method) + if result_validator: + result_validator(result) + except ValueError as exc: + return {"name": name, "method": method, "passed": False, "message": str(exc)} + return {"name": name, "method": method, "passed": True, "message": "passed"} + + +def build_conformance_report(target: str, checks: List[Dict[str, Any]]) -> Dict[str, Any]: + passed = sum(1 for check in checks if check["passed"]) + total = len(checks) + return { + "target": target, + "conformant": passed == total, + "passed": passed, + "failed": total - passed, + "total": total, + "checks": checks, + } + + +def _validate_tool_call(tool_call: Any, index: Optional[int] = None) -> None: + label = "tool call" if index is None else f"tool_calls[{index}]" + if not isinstance(tool_call, dict): + raise ValueError(f"agent/step result {label} must be an object") + if not isinstance(tool_call.get("name"), str) or not tool_call["name"]: + raise ValueError(f"agent/step result {label}.name must be a non-empty string") + if "arguments" in tool_call and not isinstance(tool_call["arguments"], dict): + raise ValueError(f"agent/step result {label}.arguments must be an object") diff --git a/runtime/python/src/ecp_runtime/manifest.py b/runtime/python/src/ecp_runtime/manifest.py index 8d6faa0..dddd207 100644 --- a/runtime/python/src/ecp_runtime/manifest.py +++ b/runtime/python/src/ecp_runtime/manifest.py @@ -1,11 +1,17 @@ from typing import Any, Dict, List, Literal, Optional import yaml -from pydantic import BaseModel, Field, model_validator +from pydantic import BaseModel, ConfigDict, Field, model_validator + + +class StrictModel(BaseModel): + """Base class that keeps runtime validation aligned with the JSON schema.""" + + model_config = ConfigDict(extra="forbid") # --- The Grader (Assertion) Schema --- -class GraderConfig(BaseModel): +class GraderConfig(StrictModel): type: Literal["text_match", "llm_judge", "tool_usage"] field: Literal["public_output", "evaluation_context", "private_thought"] = "public_output" # For text_match @@ -41,20 +47,20 @@ def _validate_by_type(self) -> "GraderConfig": return self # --- The Step (Scenario) Schema --- -class StepConfig(BaseModel): +class StepConfig(StrictModel): input: str constraints: Dict[str, Any] = Field(default_factory=dict) graders: List[GraderConfig] = Field(default_factory=list) -class ScenarioConfig(BaseModel): - name: str +class ScenarioConfig(StrictModel): + name: str = Field(min_length=1) steps: List[StepConfig] # --- The Root Manifest Schema --- -class ECPManifest(BaseModel): - manifest_version: Literal["v1"] = "v1" - name: str - target: str +class ECPManifest(StrictModel): + manifest_version: Literal["v1"] + name: str = Field(min_length=1) + target: str = Field(min_length=1) scenarios: List[ScenarioConfig] @classmethod diff --git a/runtime/python/src/ecp_runtime/runner.py b/runtime/python/src/ecp_runtime/runner.py index 97be2be..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 @@ -17,6 +17,11 @@ from urllib import error, request from urllib.parse import urlparse +from .conformance import ( + validate_initialize_result, + validate_rpc_response, + validate_step_result, +) from .graders import evaluate_step logger = logging.getLogger(__name__) @@ -67,11 +72,12 @@ def send_rpc(self, method: str, params: Dict[str, Any] = None) -> Dict[str, Any] if not params: params = {} + request_id = int(time.time() * 1000) request = { "jsonrpc": "2.0", "method": method, "params": params, - "id": int(time.time() * 1000) + "id": request_id } # Write to Agent's STDIN @@ -79,7 +85,9 @@ def send_rpc(self, method: str, params: Dict[str, Any] = None) -> Dict[str, Any] self.process.stdin.write(json_str + "\n") self.process.stdin.flush() - return self._read_json_response() + response = self._read_json_response() + _ensure_response_id(response, request_id) + return response def _read_json_response(self) -> Dict[str, Any]: start_time = time.time() @@ -168,11 +176,12 @@ def send_rpc(self, method: str, params: Dict[str, Any] = None) -> Dict[str, Any] if not params: params = {} + request_id = int(time.time() * 1000) payload = { "jsonrpc": "2.0", "method": method, "params": params, - "id": int(time.time() * 1000), + "id": request_id, } body = json.dumps(payload).encode("utf-8") req = request.Request( @@ -198,12 +207,15 @@ def send_rpc(self, method: str, params: Dict[str, Any] = None) -> Dict[str, Any] raise RuntimeError(f"HTTP RPC failed: {exc.reason}") from exc if "text/event-stream" in content_type: - return self._parse_sse_response(raw) + response = self._parse_sse_response(raw) + _ensure_response_id(response, request_id) + return response if not raw: raise RuntimeError("HTTP RPC response was empty") payload = json.loads(raw) if not isinstance(payload, dict): raise RuntimeError("HTTP RPC response must be a JSON object") + _ensure_response_id(payload, request_id) return payload def _parse_sse_response(self, raw: str) -> Dict[str, Any]: @@ -223,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 @@ -234,20 +247,35 @@ 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: + raise RuntimeError( + f"Invalid agent/initialize result at scenario='{scenario.name}': {exc}" + ) from exc scenario_steps: List[Dict[str, Any]] = [] 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") - result_data = rpc_resp.get("result", {}) + 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 step_result = StepResult( @@ -255,6 +283,7 @@ def run_scenarios(self): public_output=result_data.get("public_output"), evaluation_context=result_data.get("evaluation_context") or result_data.get("private_thought"), private_thought=result_data.get("private_thought") or result_data.get("evaluation_context"), + logs=result_data.get("logs"), tool_calls=result_data.get("tool_calls") if isinstance(result_data.get("tool_calls"), list) else None ) @@ -282,6 +311,7 @@ def run_scenarios(self): "output": step_result.public_output, "evaluation_context": step_result.evaluation_context, "tool_calls": step_result.tool_calls or [], + "logs": step_result.logs, "checks": checks }) @@ -312,20 +342,55 @@ def _ensure_rpc_success( step_idx: Optional[int], method: str, ) -> None: - if "error" not in rpc_resp: - return + where = f"scenario='{scenario_name}'" + if step_idx is not None: + where += f", step={step_idx}" + try: + validate_rpc_response(rpc_resp, method) + except ValueError as exc: + raise RuntimeError(f"RPC call failed ({method}) at {where}: {exc}") from exc - error = rpc_resp.get("error") or {} - code = error.get("code") - message = error.get("message", "Unknown JSON-RPC error") + 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}" - raise RuntimeError( - f"RPC call failed ({method}) at {where}: code={code}, message={message}" - ) + 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) return parsed.scheme in {"http", "https"} and bool(parsed.netloc) + + +def _ensure_response_id(response: Dict[str, Any], request_id: int) -> None: + response_id = response.get("id") + if type(response_id) is not type(request_id) or response_id != request_id: + raise RuntimeError( + f"JSON-RPC response id mismatch: expected {request_id}, got {response_id}" + ) diff --git a/runtime/python/src/ecp_runtime/trend.py b/runtime/python/src/ecp_runtime/trend.py index 39cd6e2..521eec4 100644 --- a/runtime/python/src/ecp_runtime/trend.py +++ b/runtime/python/src/ecp_runtime/trend.py @@ -8,10 +8,9 @@ from __future__ import annotations import json -import statistics from dataclasses import dataclass, field from pathlib import Path -from typing import List, Literal +from typing import List, Literal, Optional @dataclass @@ -81,7 +80,7 @@ def analyze(self) -> RunTrendReport: ) @staticmethod - def _load_run_point(path: Path) -> RunPoint | None: + def _load_run_point(path: Path) -> Optional[RunPoint]: try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): @@ -101,9 +100,17 @@ def _load_run_point(path: Path) -> RunPoint | None: @staticmethod def _compute_slope(pass_rates: List[float]) -> float: - x = list(range(len(pass_rates))) - slope, _intercept = statistics.linear_regression(x, pass_rates) - return slope + count = len(pass_rates) + if count < 2: + return 0.0 + mean_x = (count - 1) / 2 + mean_y = sum(pass_rates) / count + numerator = sum( + (index - mean_x) * (rate - mean_y) + for index, rate in enumerate(pass_rates) + ) + denominator = sum((index - mean_x) ** 2 for index in range(count)) + return numerator / denominator if denominator else 0.0 def _classify(self, slope: float) -> Literal["improving", "degrading", "stable"]: if slope > self._IMPROVING_THRESHOLD: diff --git a/runtime/python/tests/test_cli.py b/runtime/python/tests/test_cli.py index 4d51d54..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]) @@ -95,6 +116,61 @@ def test_init_command_creates_starter_files(self) -> None: self.assertTrue(manifest.exists()) self.assertIn("evaluation_context", manifest.read_text(encoding="utf-8")) + def test_conformance_json_report(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", "public_output": "ok"}}, + {"jsonrpc": "2.0", "id": 3, "result": True}, + ] + fake_runtime = mock.Mock() + fake_runtime._create_agent.return_value = fake_agent + + with mock.patch("ecp_runtime.cli.ECPRunner", return_value=fake_runtime): + result = self.runner.invoke(app, ["conformance", "--target", "python agent.py", "--json"]) + + self.assertEqual(result.exit_code, 0, msg=result.output) + payload = json.loads(result.stdout) + self.assertTrue(payload["conformant"]) + 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 = [ + {"jsonrpc": "2.0", "id": 1, "result": {}}, + {"jsonrpc": "2.0", "id": 2, "result": {"status": "invalid"}}, + {"jsonrpc": "2.0", "id": 3, "result": True}, + ] + fake_runtime = mock.Mock() + fake_runtime._create_agent.return_value = fake_agent + + with mock.patch("ecp_runtime.cli.ECPRunner", return_value=fake_runtime): + result = self.runner.invoke(app, ["conformance", "--target", "python agent.py"]) + + self.assertEqual(result.exit_code, 1, msg=result.output) + self.assertIn("FAIL | step result contract", result.output) + if __name__ == "__main__": unittest.main() diff --git a/runtime/python/tests/test_conformance.py b/runtime/python/tests/test_conformance.py new file mode 100644 index 0000000..5bcbd1d --- /dev/null +++ b/runtime/python/tests/test_conformance.py @@ -0,0 +1,91 @@ +import sys +import unittest +from pathlib import Path + +RUNTIME_SRC = Path(__file__).resolve().parents[1] / "src" +if str(RUNTIME_SRC) not in sys.path: + sys.path.insert(0, str(RUNTIME_SRC)) + +from ecp_runtime.conformance import ( + build_conformance_report, + conformance_check, + validate_initialize_result, + validate_reset_result, + validate_rpc_response, + validate_step_result, +) + + +class ConformanceTests(unittest.TestCase): + def test_valid_step_result(self) -> None: + result = { + "status": "done", + "public_output": "ok", + "evaluation_context": "verified", + "tool_calls": [{"name": "lookup", "arguments": {"id": 1}}], + "logs": "complete", + } + + self.assertIs(validate_step_result(result), result) + + def test_initialize_contract(self) -> None: + result = {"name": "agent", "capabilities": {}} + self.assertIs(validate_initialize_result(result), result) + + with self.assertRaisesRegex(ValueError, "capabilities"): + validate_initialize_result({"name": "agent"}) + + def test_reset_contract(self) -> None: + self.assertTrue(validate_reset_result(True)) + with self.assertRaisesRegex(ValueError, "must be true"): + validate_reset_result(None) + + def test_invalid_status_is_rejected(self) -> None: + with self.assertRaisesRegex(ValueError, "status"): + validate_step_result({"status": "complete"}) + + def test_invalid_tool_call_is_rejected(self) -> None: + with self.assertRaisesRegex(ValueError, "name"): + validate_step_result({"status": "done", "tool_calls": [{"arguments": {}}]}) + + def test_rpc_envelope_requires_id(self) -> None: + with self.assertRaisesRegex(ValueError, "include id"): + validate_rpc_response({"jsonrpc": "2.0", "result": {}}, "agent/initialize") + + def test_rpc_envelope_rejects_result_and_error(self) -> None: + with self.assertRaisesRegex(ValueError, "both result and error"): + validate_rpc_response( + { + "jsonrpc": "2.0", + "id": 1, + "result": {}, + "error": {"code": -32000, "message": "failure"}, + }, + "agent/initialize", + ) + + def test_report_has_stable_counts(self) -> None: + checks = [ + conformance_check( + "initialize", + "agent/initialize", + {"jsonrpc": "2.0", "id": 1, "result": {}}, + ), + conformance_check( + "step", + "agent/step", + {"jsonrpc": "2.0", "id": 2, "result": {"status": "invalid"}}, + result_validator=validate_step_result, + ), + ] + + report = build_conformance_report("python agent.py", checks) + + self.assertFalse(report["conformant"]) + self.assertEqual(report["passed"], 1) + self.assertEqual(report["failed"], 1) + self.assertEqual(report["total"], 2) + + +if __name__ == "__main__": + unittest.main() 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_manifest.py b/runtime/python/tests/test_manifest.py index 0563d90..3ab721a 100644 --- a/runtime/python/tests/test_manifest.py +++ b/runtime/python/tests/test_manifest.py @@ -29,6 +29,9 @@ def test_llm_judge_requires_prompt(self) -> None: with self.assertRaises(ValidationError): GraderConfig(type="llm_judge") + with self.assertRaises(ValidationError): + GraderConfig(type="llm_judge", prompt=" ") + def test_invalid_grader_type_rejected(self) -> None: with self.assertRaises(ValidationError): GraderConfig(type="unknown") # type: ignore[arg-type] @@ -51,6 +54,25 @@ def test_manifest_root_must_be_mapping(self) -> None: with self.assertRaises(ValueError): ECPManifest.from_yaml(tmp_path) + def test_unknown_manifest_fields_are_rejected(self) -> None: + with self.assertRaises(ValidationError): + ECPManifest( + manifest_version="v1", + name="test", + target="python agent.py", + scenarios=[], + unexpected=True, + ) + + def test_unknown_grader_fields_are_rejected(self) -> None: + with self.assertRaises(ValidationError): + GraderConfig( + type="text_match", + condition="contains", + value="ok", + unexpected=True, + ) + if __name__ == "__main__": unittest.main() diff --git a/runtime/python/tests/test_runner.py b/runtime/python/tests/test_runner.py index f9ee956..a937540 100644 --- a/runtime/python/tests/test_runner.py +++ b/runtime/python/tests/test_runner.py @@ -1,3 +1,4 @@ +import json import sys import unittest from pathlib import Path @@ -9,7 +10,12 @@ sys.path.insert(0, str(RUNTIME_SRC)) from ecp_runtime.manifest import StepConfig -from ecp_runtime.runner import ECPRunner, HTTPAgentClient +from ecp_runtime.runner import ( + ECPRunner, + HTTPAgentClient, + _ensure_response_id, + resolve_rpc_timeout, +) class RunnerTests(unittest.TestCase): @@ -73,6 +79,24 @@ def send_rpc(self, method, params=None): self.assertIn("step=1", msg) self.assertIn("boom", msg) + def test_runner_rejects_invalid_initialize_result(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): + return {"jsonrpc": "2.0", "id": 1, "result": {"name": "missing-capabilities"}} + + with mock.patch("ecp_runtime.runner.AgentProcess", FakeAgentProcess): + with self.assertRaisesRegex(RuntimeError, "Invalid agent/initialize result"): + ECPRunner(self._manifest()).run_scenarios() + def test_runner_uses_http_client_for_url_target(self) -> None: runner = ECPRunner(SimpleNamespace(target="http://127.0.0.1:8765/ecp", scenarios=[])) @@ -82,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"} @@ -102,6 +186,10 @@ def fake_urlopen(req, timeout): captured["timeout"] = timeout captured["body"] = req.data captured["accept"] = req.headers.get("Accept") + request_id = json.loads(req.data.decode("utf-8"))["id"] + FakeResponse.read = lambda self: json.dumps( + {"jsonrpc": "2.0", "id": request_id, "result": {"ok": True}} + ).encode("utf-8") return FakeResponse() with mock.patch("ecp_runtime.runner.request.urlopen", fake_urlopen): @@ -117,6 +205,13 @@ def fake_urlopen(req, timeout): self.assertIn('"method": "agent/step"', body) self.assertIn('"input": "hi"', body) + def test_response_id_must_match_request(self) -> None: + with self.assertRaisesRegex(RuntimeError, "id mismatch"): + _ensure_response_id({"jsonrpc": "2.0", "id": 2, "result": {}}, 1) + + with self.assertRaisesRegex(RuntimeError, "id mismatch"): + _ensure_response_id({"jsonrpc": "2.0", "id": True, "result": {}}, 1) + if __name__ == "__main__": unittest.main() diff --git a/runtime/python/tests/test_trend.py b/runtime/python/tests/test_trend.py index c7ed307..ce0078e 100644 --- a/runtime/python/tests/test_trend.py +++ b/runtime/python/tests/test_trend.py @@ -65,6 +65,9 @@ def test_single_run_returns_stable(self) -> None: self.assertEqual(report.direction, "stable") self.assertEqual(report.pass_rate_slope, 0.0) + def test_slope_matches_expected_least_squares_result(self) -> None: + self.assertAlmostEqual(RunTrendAnalyzer._compute_slope([0.1, 0.5, 0.9]), 0.4) + class RunTrendAnalyzerFileTests(unittest.TestCase): def test_load_run_point_from_file(self) -> None: diff --git a/schema/agent-result.schema.json b/schema/agent-result.schema.json index d3bb1d6..baa12a2 100644 --- a/schema/agent-result.schema.json +++ b/schema/agent-result.schema.json @@ -26,6 +26,10 @@ "items": { "$ref": "./tool-call.schema.json" } + }, + "logs": { + "type": ["string", "null"], + "description": "Optional evaluator-visible execution logs." } }, "required": ["status"] diff --git a/schema/manifest.schema.json b/schema/manifest.schema.json index 2d39c71..bdd33af 100644 --- a/schema/manifest.schema.json +++ b/schema/manifest.schema.json @@ -44,7 +44,7 @@ "type": "array", "items": { "type": "object", - "additionalProperties": true, + "additionalProperties": false, "properties": { "type": { "type": "string", @@ -53,9 +53,57 @@ "field": { "type": "string", "enum": ["public_output", "evaluation_context", "private_thought"] + }, + "condition": { + "type": ["string", "null"], + "enum": ["contains", "equals", "does_not_contain", "regex", null] + }, + "value": {"type": ["string", "null"]}, + "pattern": {"type": ["string", "null"], "minLength": 1}, + "prompt": { + "type": ["string", "null"], + "minLength": 1, + "pattern": "\\S" + }, + "assertion": {"type": ["string", "null"]}, + "tool_name": {"type": ["string", "null"]}, + "arguments": { + "type": "object", + "additionalProperties": true } }, - "required": ["type"] + "required": ["type"], + "allOf": [ + { + "if": {"properties": {"type": {"const": "text_match"}}}, + "then": { + "required": ["condition"], + "properties": {"condition": {"type": "string"}}, + "allOf": [ + { + "if": {"properties": {"condition": {"const": "regex"}}}, + "then": { + "required": ["pattern"], + "properties": {"pattern": {"type": "string", "minLength": 1}} + }, + "else": { + "required": ["value"], + "properties": {"value": {"type": "string"}} + } + } + ] + } + }, + { + "if": {"properties": {"type": {"const": "llm_judge"}}}, + "then": { + "required": ["prompt"], + "properties": { + "prompt": {"type": "string", "minLength": 1, "pattern": "\\S"} + } + } + } + ] } } }, diff --git a/schema/tool-call.schema.json b/schema/tool-call.schema.json index 29072a3..9f7ecd3 100644 --- a/schema/tool-call.schema.json +++ b/schema/tool-call.schema.json @@ -6,7 +6,8 @@ "additionalProperties": true, "properties": { "name": { - "type": "string" + "type": "string", + "minLength": 1 }, "arguments": { "type": "object", 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/decorators.py b/sdk/python/src/ecp/decorators.py index f0901df..9c9a198 100644 --- a/sdk/python/src/ecp/decorators.py +++ b/sdk/python/src/ecp/decorators.py @@ -15,6 +15,22 @@ class Result: logs: Optional[str] = None def __post_init__(self): + if self.status not in {"done", "paused"}: + raise ValueError("status must be 'done' or 'paused'") + for field_name in ("public_output", "evaluation_context", "private_thought", "logs"): + value = getattr(self, field_name) + if value is not None and not isinstance(value, str): + raise TypeError(f"{field_name} must be a string or None") + if self.tool_calls is not None: + if not isinstance(self.tool_calls, list): + raise TypeError("tool_calls must be a list or None") + for index, tool_call in enumerate(self.tool_calls): + if not isinstance(tool_call, dict): + raise TypeError(f"tool_calls[{index}] must be a dictionary") + if not isinstance(tool_call.get("name"), str) or not tool_call["name"]: + raise ValueError(f"tool_calls[{index}].name must be a non-empty string") + if "arguments" in tool_call and not isinstance(tool_call["arguments"], dict): + raise TypeError(f"tool_calls[{index}].arguments must be a dictionary") if self.evaluation_context is None and self.private_thought is not None: self.evaluation_context = self.private_thought elif self.private_thought is None and self.evaluation_context is not None: diff --git a/sdk/python/src/ecp/server.py b/sdk/python/src/ecp/server.py index 5f09bb1..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): @@ -194,17 +260,22 @@ def _handle_step(params): "public_output": result.public_output, "evaluation_context": result.evaluation_context, "private_thought": result.private_thought, - "tool_calls": result.tool_calls + "tool_calls": result.tool_calls, + "logs": result.logs, } 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 e844356..7e4b308 100644 --- a/sdk/python/tests/test_server.py +++ b/sdk/python/tests/test_server.py @@ -1,8 +1,10 @@ +import asyncio import json import sys import threading import unittest from pathlib import Path +from unittest import mock from urllib import error, request SDK_SRC = Path(__file__).resolve().parents[1] / "src" @@ -10,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") @@ -20,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() @@ -37,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( @@ -73,6 +96,25 @@ def test_result_syncs_private_thought_alias(self) -> None: self.assertEqual(result.evaluation_context, "legacy") + def test_result_rejects_invalid_status(self) -> None: + with self.assertRaisesRegex(ValueError, "status"): + Result(status="complete", public_output="ok") + + def test_result_rejects_invalid_tool_call(self) -> None: + with self.assertRaisesRegex(ValueError, "name"): + Result(tool_calls=[{"arguments": {}}]) + + def test_server_serializes_logs(self) -> None: + result = Result(public_output="ok", logs="trace") + + with mock.patch.object(HTTPTestAgent, "step", return_value=result): + status, body, _headers = self._post( + {"jsonrpc": "2.0", "id": 1, "method": "agent/step", "params": {"input": "hello"}} + ) + + self.assertEqual(status, 200) + self.assertEqual(body["result"]["logs"], "trace") + def test_post_json_rpc_notification_returns_accepted(self) -> None: status, body, _headers = self._post( {"jsonrpc": "2.0", "method": "agent/initialize", "params": {}} @@ -102,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() diff --git a/spec/protocol.md b/spec/protocol.md index b7326c4..9a1af47 100644 --- a/spec/protocol.md +++ b/spec/protocol.md @@ -70,6 +70,7 @@ The reference Python SDK currently returns JSON responses for `POST` requests an - `evaluation_context` (string | null): evaluator-safe audit context, reasoning summary, policy evidence, or trace summary. - `private_thought` (string | null): deprecated compatibility alias for `evaluation_context`. - `tool_calls` (array | null): tools the agent invoked. +- `logs` (string | null): optional evaluator-visible execution logs. ECP does not require raw chain-of-thought. New agents should use `evaluation_context` for concise evaluator-safe evidence. @@ -143,4 +144,3 @@ JSON Schemas live in `schema/`. Protocol implementers can run: ```bash ecp conformance --target "python examples/customer_support_demo/agent.py" ``` -