feat: add ACP bridge for codeg integration
Stdio-based JSON-RPC bridge implementing the Agent Client Protocol, allowing codeg to connect to GenericAgent as a live ACP agent. Handles Windows binary stdout/stdin, redirects stray prints to stderr, and converts GenericAgent's cumulative text queue to ACP delta streaming. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
yiqi-017andClaude Opus 4.6 committed
1 parent
d626c9f02c
commit
be4d2f58c1
1 file changed
+349
@@ -0,0 +1,349 @@
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
# Must run BEFORE importing agentmain — it reconfigures stdout at import time,
|
||||
# and its submodules may print() during init. We capture the raw binary stdout
|
||||
# for ACP JSON-RPC, then redirect the text-mode stdout to stderr so any stray
|
||||
# prints from agentmain/llmcore don't pollute the ACP channel.
|
||||
if sys.platform == "win32":
|
||||
import msvcrt
|
||||
_stdout_fd = os.dup(sys.__stdout__.fileno())
|
||||
msvcrt.setmode(_stdout_fd, os.O_BINARY)
|
||||
_acp_stdout = os.fdopen(_stdout_fd, "wb", buffering=0)
|
||||
msvcrt.setmode(sys.stdin.fileno(), os.O_BINARY)
|
||||
# Redirect the original stdout fd to stderr so child processes
|
||||
# (tool calls) don't write into the ACP JSON-RPC channel.
|
||||
os.dup2(sys.stderr.fileno(), sys.__stdout__.fileno())
|
||||
else:
|
||||
_acp_stdout = sys.__stdout__.buffer if hasattr(sys.__stdout__, 'buffer') else sys.__stdout__
|
||||
os.dup2(sys.stderr.fileno(), sys.__stdout__.fileno())
|
||||
|
||||
|
||||
class _StdoutToStderrRouter(io.TextIOBase):
|
||||
"""Redirect text-mode stdout to stderr so agentmain prints don't leak."""
|
||||
def writable(self): return True
|
||||
def write(self, s):
|
||||
if s:
|
||||
sys.stderr.write(s)
|
||||
sys.stderr.flush()
|
||||
return len(s) if s else 0
|
||||
def flush(self): sys.stderr.flush()
|
||||
|
||||
sys.stdout = _StdoutToStderrRouter()
|
||||
|
||||
import argparse
|
||||
import queue
|
||||
import threading
|
||||
import traceback
|
||||
import uuid
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from agentmain import GeneraticAgent
|
||||
|
||||
|
||||
JSONRPC_VERSION = "2.0"
|
||||
ACP_PROTOCOL_VERSION = 1
|
||||
|
||||
|
||||
def eprint(*args: Any) -> None:
|
||||
print(*args, file=sys.stderr, flush=True)
|
||||
|
||||
|
||||
def make_text_block(text: str) -> Dict[str, Any]:
|
||||
return {"type": "text", "text": text}
|
||||
|
||||
|
||||
def make_session_update(session_id: str, update: Dict[str, Any]) -> Dict[str, Any]:
|
||||
return {
|
||||
"jsonrpc": JSONRPC_VERSION,
|
||||
"method": "session/update",
|
||||
"params": {"sessionId": session_id, "update": update},
|
||||
}
|
||||
|
||||
|
||||
def compact_json(obj: Dict[str, Any]) -> str:
|
||||
return json.dumps(obj, ensure_ascii=False, separators=(",", ":"))
|
||||
|
||||
|
||||
def parse_jsonrpc_line(line: str) -> Optional[Dict[str, Any]]:
|
||||
stripped = line.strip()
|
||||
if not stripped:
|
||||
return None
|
||||
try:
|
||||
obj = json.loads(stripped)
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
return obj if isinstance(obj, dict) else None
|
||||
|
||||
|
||||
def content_blocks_to_text(blocks: List[Dict[str, Any]]) -> str:
|
||||
parts: List[str] = []
|
||||
for block in blocks:
|
||||
if not isinstance(block, dict):
|
||||
continue
|
||||
block_type = block.get("type")
|
||||
if block_type == "text":
|
||||
text = block.get("text")
|
||||
if isinstance(text, str) and text:
|
||||
parts.append(text)
|
||||
elif block_type == "resource_link":
|
||||
name = block.get("name") or "resource"
|
||||
uri = block.get("uri") or ""
|
||||
desc = block.get("description") or ""
|
||||
parts.append(f"[ResourceLink] {name}: {uri}\n{desc}".strip())
|
||||
elif block_type == "resource":
|
||||
uri = block.get("uri") or "resource"
|
||||
text = block.get("text")
|
||||
if isinstance(text, str) and text:
|
||||
parts.append(f"[Resource] {uri}\n{text}")
|
||||
else:
|
||||
parts.append(f"[Resource] {uri}")
|
||||
elif block_type == "image":
|
||||
uri = block.get("uri") or "inline-image"
|
||||
parts.append(f"[Image omitted] {uri}")
|
||||
else:
|
||||
parts.append(f"[Unsupported content block: {block_type}]")
|
||||
return "\n\n".join(p for p in parts if p).strip()
|
||||
|
||||
|
||||
def jsonrpc_error(code: int, message: str, req_id: Any = None, data: Any = None) -> Dict[str, Any]:
|
||||
err: Dict[str, Any] = {"code": code, "message": message}
|
||||
if data is not None:
|
||||
err["data"] = data
|
||||
return {"jsonrpc": JSONRPC_VERSION, "id": req_id, "error": err}
|
||||
|
||||
|
||||
def jsonrpc_result(req_id: Any, result: Any) -> Dict[str, Any]:
|
||||
return {"jsonrpc": JSONRPC_VERSION, "id": req_id, "result": result}
|
||||
|
||||
|
||||
@dataclass
|
||||
class SessionState:
|
||||
session_id: str
|
||||
cwd: str
|
||||
agent: GeneraticAgent
|
||||
current_prompt_id: Any = None
|
||||
prompt_lock: threading.Lock = field(default_factory=threading.Lock)
|
||||
|
||||
|
||||
class GenericAgentAcpBridge:
|
||||
def __init__(self, llm_no: int = 0):
|
||||
self.llm_no = llm_no
|
||||
self._json_out = _acp_stdout
|
||||
self._write_lock = threading.Lock()
|
||||
self._sessions: Dict[str, SessionState] = {}
|
||||
self._shutdown = False
|
||||
|
||||
def write_message(self, msg: Dict[str, Any]) -> None:
|
||||
payload = compact_json(msg)
|
||||
raw = (payload + "\n").encode("utf-8")
|
||||
method = msg.get("method", msg.get("id", "?"))
|
||||
eprint(f"[ACP-BRIDGE] >>> {payload[:500]}")
|
||||
try:
|
||||
with self._write_lock:
|
||||
self._json_out.write(raw)
|
||||
self._json_out.flush()
|
||||
except Exception as e:
|
||||
eprint(f"[ACP-BRIDGE] WRITE FAILED: {type(e).__name__}: {e}")
|
||||
|
||||
def new_agent(self) -> GeneraticAgent:
|
||||
agent = GeneraticAgent()
|
||||
agent.next_llm(self.llm_no)
|
||||
agent.verbose = False
|
||||
threading.Thread(target=agent.run, daemon=True).start()
|
||||
return agent
|
||||
|
||||
def handle_initialize(self, req_id: Any, params: Dict[str, Any]) -> None:
|
||||
requested_version = params.get("protocolVersion", ACP_PROTOCOL_VERSION)
|
||||
version = ACP_PROTOCOL_VERSION if requested_version == ACP_PROTOCOL_VERSION else ACP_PROTOCOL_VERSION
|
||||
result = {
|
||||
"protocolVersion": version,
|
||||
"agentCapabilities": {
|
||||
"loadSession": False,
|
||||
"mcpCapabilities": {"http": False, "sse": False},
|
||||
"promptCapabilities": {
|
||||
"image": False,
|
||||
"audio": False,
|
||||
"embeddedContext": False,
|
||||
},
|
||||
"sessionCapabilities": {},
|
||||
},
|
||||
"agentInfo": {
|
||||
"name": "genericagent-acp",
|
||||
"title": "GenericAgent",
|
||||
"version": "0.1.0",
|
||||
},
|
||||
"authMethods": [],
|
||||
}
|
||||
self.write_message(jsonrpc_result(req_id, result))
|
||||
|
||||
def handle_session_new(self, req_id: Any, params: Dict[str, Any]) -> None:
|
||||
cwd = params.get("cwd")
|
||||
if not isinstance(cwd, str) or not cwd:
|
||||
self.write_message(jsonrpc_error(-32602, "cwd is required", req_id))
|
||||
return
|
||||
if not os.path.isabs(cwd):
|
||||
cwd = os.path.abspath(cwd)
|
||||
session_id = f"ga_{uuid.uuid4().hex}"
|
||||
agent = self.new_agent()
|
||||
session = SessionState(session_id=session_id, cwd=cwd, agent=agent)
|
||||
self._sessions[session_id] = session
|
||||
self.write_message(
|
||||
jsonrpc_result(
|
||||
req_id,
|
||||
{
|
||||
"sessionId": session_id,
|
||||
"modes": None,
|
||||
"configOptions": None,
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
def handle_session_prompt(self, req_id: Any, params: Dict[str, Any]) -> None:
|
||||
session_id = params.get("sessionId")
|
||||
prompt_blocks = params.get("prompt")
|
||||
session = self._sessions.get(session_id)
|
||||
if session is None:
|
||||
self.write_message(jsonrpc_error(-32602, "unknown sessionId", req_id))
|
||||
return
|
||||
if not isinstance(prompt_blocks, list):
|
||||
self.write_message(jsonrpc_error(-32602, "prompt must be an array", req_id))
|
||||
return
|
||||
prompt_text = content_blocks_to_text(prompt_blocks)
|
||||
if not prompt_text:
|
||||
self.write_message(jsonrpc_error(-32602, "prompt must contain text or supported content", req_id))
|
||||
return
|
||||
|
||||
with session.prompt_lock:
|
||||
if session.current_prompt_id is not None:
|
||||
self.write_message(
|
||||
jsonrpc_error(-32603, "session already has an active prompt", req_id)
|
||||
)
|
||||
return
|
||||
session.current_prompt_id = req_id
|
||||
|
||||
def run_prompt() -> None:
|
||||
stop_reason = "end_turn"
|
||||
try:
|
||||
dq = session.agent.put_task(prompt_text, source="acp")
|
||||
self._drain_agent_queue(session, dq)
|
||||
except Exception as exc:
|
||||
stop_reason = "end_turn"
|
||||
self.write_message(
|
||||
make_session_update(
|
||||
session.session_id,
|
||||
{
|
||||
"sessionUpdate": "agent_message_chunk",
|
||||
"content": make_text_block(
|
||||
f"[Bridge error] {type(exc).__name__}: {exc}"
|
||||
),
|
||||
},
|
||||
)
|
||||
)
|
||||
eprint("[GenericAgent ACP] prompt thread failed:", traceback.format_exc())
|
||||
finally:
|
||||
with session.prompt_lock:
|
||||
finished_req_id = session.current_prompt_id
|
||||
session.current_prompt_id = None
|
||||
if finished_req_id is not None:
|
||||
import time
|
||||
time.sleep(0.1)
|
||||
self.write_message(
|
||||
jsonrpc_result(finished_req_id, {"stopReason": stop_reason})
|
||||
)
|
||||
|
||||
threading.Thread(target=run_prompt, daemon=True).start()
|
||||
|
||||
def _drain_agent_queue(self, session: SessionState, dq: "queue.Queue[Dict[str, Any]]") -> None:
|
||||
last_sent = ""
|
||||
while True:
|
||||
item = dq.get()
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
text = item.get("done") or item.get("next") or ""
|
||||
if isinstance(text, str) and len(text) > len(last_sent):
|
||||
delta = text[len(last_sent):]
|
||||
last_sent = text
|
||||
try:
|
||||
self.write_message(
|
||||
make_session_update(
|
||||
session.session_id,
|
||||
{
|
||||
"sessionUpdate": "agent_message_chunk",
|
||||
"content": make_text_block(delta),
|
||||
},
|
||||
)
|
||||
)
|
||||
except Exception as e:
|
||||
eprint(f"[ACP-BRIDGE] ERROR writing update: {e}")
|
||||
if "done" in item:
|
||||
break
|
||||
|
||||
def handle_session_cancel(self, params: Dict[str, Any]) -> None:
|
||||
session_id = params.get("sessionId")
|
||||
session = self._sessions.get(session_id)
|
||||
if session is None:
|
||||
return
|
||||
if session.current_prompt_id is not None:
|
||||
session.agent.abort()
|
||||
|
||||
def handle_message(self, msg: Dict[str, Any]) -> None:
|
||||
method = msg.get("method")
|
||||
req_id = msg.get("id")
|
||||
params = msg.get("params") or {}
|
||||
|
||||
try:
|
||||
if method == "initialize":
|
||||
self.handle_initialize(req_id, params)
|
||||
elif method == "session/new":
|
||||
self.handle_session_new(req_id, params)
|
||||
elif method == "session/prompt":
|
||||
self.handle_session_prompt(req_id, params)
|
||||
elif method == "session/cancel":
|
||||
self.handle_session_cancel(params)
|
||||
elif method == "session/load":
|
||||
self.write_message(jsonrpc_error(-32601, "session/load not supported", req_id))
|
||||
elif method == "session/list":
|
||||
self.write_message(jsonrpc_error(-32601, "session/list not supported", req_id))
|
||||
elif method == "session/close":
|
||||
self.write_message(jsonrpc_result(req_id, {}))
|
||||
elif method is None:
|
||||
if req_id is not None:
|
||||
self.write_message(jsonrpc_error(-32600, "invalid request", req_id))
|
||||
else:
|
||||
if req_id is not None:
|
||||
self.write_message(jsonrpc_error(-32601, f"method not found: {method}", req_id))
|
||||
except Exception as exc:
|
||||
eprint("[GenericAgent ACP] request handler failed:", traceback.format_exc())
|
||||
if req_id is not None:
|
||||
self.write_message(
|
||||
jsonrpc_error(-32603, f"internal error: {type(exc).__name__}: {exc}", req_id)
|
||||
)
|
||||
|
||||
def serve(self) -> None:
|
||||
eprint("[GenericAgent ACP] bridge started")
|
||||
stdin = io.TextIOWrapper(sys.stdin.buffer, encoding="utf-8", errors="replace") if hasattr(sys.stdin, 'buffer') else sys.stdin
|
||||
for raw_line in stdin:
|
||||
msg = parse_jsonrpc_line(raw_line)
|
||||
if msg is None:
|
||||
continue
|
||||
self.handle_message(msg)
|
||||
if self._shutdown:
|
||||
break
|
||||
eprint("[GenericAgent ACP] bridge stopped")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="GenericAgent ACP bridge over stdio")
|
||||
parser.add_argument("--llm-no", type=int, default=0, help="LLM index for GenericAgent")
|
||||
args = parser.parse_args()
|
||||
bridge = GenericAgentAcpBridge(llm_no=args.llm_no)
|
||||
bridge.serve()
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in new issue
Block a user