forked from cleveragents/cleveragents-core
freemo
680cc8c352
Add A2aStdioTransport class that manages subprocess lifecycle and communicates via newline-delimited JSON-RPC 2.0 messages over stdin/stdout. Add A2aTransport protocol (typing.Protocol) defining the abstract transport interface. Add A2aTransportSelector that picks A2aStdioTransport in local mode (no server URL) or A2aHttpTransport in server mode. New files: - transport_protocol.py: A2aTransport Protocol - stdio_transport.py: A2aStdioTransport with subprocess management - transport_selector.py: A2aTransportSelector configuration-based - echo_agent.py: Minimal echo agent fixture for tests Tests: - 20 Behave BDD scenarios covering protocol conformance, construction, subprocess lifecycle, send/receive, context manager, transport selector, and contract round-trip tests - 7 Robot Framework integration tests with helper script ISSUES CLOSED: #691
59 lines
1.5 KiB
Python
59 lines
1.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Minimal echo agent for A2A stdio transport tests.
|
|
|
|
Reads newline-delimited JSON-RPC requests from stdin and writes
|
|
JSON-RPC responses to stdout. Each request is echoed back as a
|
|
successful response with the original params in the result.
|
|
|
|
This is used by Behave and Robot Framework tests to verify the
|
|
stdio transport without needing the full CleverAgents agent.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
|
|
|
|
def main() -> None:
|
|
"""Read JSON-RPC requests from stdin, write responses to stdout."""
|
|
for line in sys.stdin:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
|
|
try:
|
|
request = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
error_response = {
|
|
"jsonrpc": "2.0",
|
|
"id": None,
|
|
"error": {
|
|
"code": -32700,
|
|
"message": "Parse error",
|
|
},
|
|
}
|
|
sys.stdout.write(json.dumps(error_response) + "\n")
|
|
sys.stdout.flush()
|
|
continue
|
|
|
|
request_id = request.get("id")
|
|
method = request.get("method", "")
|
|
params = request.get("params") or {}
|
|
|
|
response = {
|
|
"jsonrpc": "2.0",
|
|
"id": request_id,
|
|
"result": {
|
|
"method": method,
|
|
"params": params,
|
|
"echo": True,
|
|
},
|
|
}
|
|
sys.stdout.write(json.dumps(response) + "\n")
|
|
sys.stdout.flush()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|