342 lines
12 KiB
Python
342 lines
12 KiB
Python
"""``POST /v1/messages`` frontend.
|
|
|
|
Exposes the gateway as an Anthropic-compatible Messages endpoint, so any
|
|
client that already speaks Anthropic (Cursor, Cline, the official SDK,
|
|
``curl``) can hit a configured agent by passing its name as ``model``.
|
|
|
|
A Claude agent behind this endpoint is a ``deep`` conversation: the client
|
|
knows nothing about our ids, so the text fingerprint of the history it
|
|
sends is the ``(anthropic, fingerprint)`` binding of the conversation,
|
|
rebound after every turn to the fingerprint the next request will carry.
|
|
A history nobody has seen becomes a new conversation, materialized by the
|
|
home frontend of ``deep`` (the vault file), and every reply is published
|
|
on the bus so that file follows the chat. Other agents (Raycast) stay
|
|
stateless and are only archived through ``turn_log_handlers``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import uuid
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from fastapi import FastAPI, HTTPException, Request, status
|
|
from fastapi.responses import JSONResponse, StreamingResponse
|
|
|
|
from beaver_gateway.agents.claude import ClaudeAgent
|
|
from beaver_gateway.core import audit
|
|
from beaver_gateway.core.transcript import fingerprint, text_of
|
|
from beaver_gateway.core.turn_capture import TurnCapture
|
|
from beaver_gateway.core.turn_record import TurnRecord
|
|
from beaver_gateway.frontends._accumulate import StreamAccumulator
|
|
from beaver_gateway.frontends._auth import require_token
|
|
from beaver_gateway.frontends.base import Frontend
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import AsyncIterator, Awaitable, Callable
|
|
|
|
from anthropic.types import Message, MessageParam
|
|
|
|
from beaver_gateway.core.conversations import Conversations
|
|
from beaver_gateway.core.events import MessageStreamEvent
|
|
from beaver_gateway.frontends.base import GatewayRuntime
|
|
from beaver_gateway.storage.models import Conversation
|
|
|
|
|
|
_log = logging.getLogger("beaver_gateway.frontends.anthropic")
|
|
|
|
|
|
__all__ = ["AnthropicMessagesFrontend"]
|
|
|
|
FRONTEND = "anthropic"
|
|
_TITLE_LEN = 80
|
|
|
|
|
|
class AnthropicMessagesFrontend(Frontend):
|
|
"""FastAPI app behind ``POST /v1/messages`` + ``GET /v1/models``."""
|
|
|
|
name = FRONTEND
|
|
kinds = ("deep",)
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
host: str = "0.0.0.0", # noqa: S104
|
|
port: int = 8000,
|
|
public_base_url: str | None = None,
|
|
) -> None:
|
|
self.host = host
|
|
self.port = port
|
|
self.public_base_url = public_base_url.rstrip("/") if public_base_url else None
|
|
self._runtime: GatewayRuntime | None = None
|
|
self._app: FastAPI | None = None
|
|
|
|
def configure(self, runtime: GatewayRuntime) -> None:
|
|
self._runtime = runtime
|
|
self._app = self._build_app(runtime)
|
|
|
|
async def serve(self) -> None:
|
|
import uvicorn
|
|
|
|
if self._app is None:
|
|
msg = "configure() must be called before serve()"
|
|
raise RuntimeError(msg)
|
|
|
|
config = uvicorn.Config(
|
|
self._app, host=self.host, port=self.port, log_level="info"
|
|
)
|
|
server = uvicorn.Server(config)
|
|
await server.serve()
|
|
|
|
def _build_app(self, runtime: GatewayRuntime) -> FastAPI:
|
|
app = FastAPI(title="beaver-gateway / Anthropic Messages")
|
|
|
|
@app.get("/healthz")
|
|
async def healthz() -> dict[str, str]:
|
|
return {"status": "ok"}
|
|
|
|
@app.get("/v1/models")
|
|
async def list_models(request: Request) -> dict[str, Any]:
|
|
await require_token(request, runtime, scope="messages")
|
|
data = [
|
|
{
|
|
"type": "model",
|
|
"id": a.name,
|
|
"display_name": a.name,
|
|
"created_at": None,
|
|
}
|
|
for a in runtime.agents
|
|
]
|
|
return {"data": data, "has_more": False, "first_id": None, "last_id": None}
|
|
|
|
@app.post("/v1/messages")
|
|
async def create_message(request: Request) -> Any:
|
|
token_name = await require_token(request, runtime, scope="messages")
|
|
try:
|
|
body = await request.json()
|
|
except json.JSONDecodeError as exc:
|
|
raise HTTPException(
|
|
status.HTTP_400_BAD_REQUEST, f"invalid JSON: {exc}"
|
|
) from exc
|
|
|
|
model = body.get("model")
|
|
if not isinstance(model, str):
|
|
raise HTTPException(
|
|
status.HTTP_400_BAD_REQUEST, "missing or non-string `model`"
|
|
)
|
|
|
|
agent = runtime.agents.get(model)
|
|
if agent is None:
|
|
raise HTTPException(
|
|
status.HTTP_404_NOT_FOUND, f"unknown agent: {model!r}"
|
|
)
|
|
|
|
backend = runtime.backends.get(agent.name)
|
|
if backend is None:
|
|
raise HTTPException(
|
|
status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
f"no backend configured for agent {agent.name!r}",
|
|
)
|
|
|
|
messages = body.get("messages") or []
|
|
system = body.get("system")
|
|
system_str = system if isinstance(system, str) else None
|
|
stream_flag = bool(body.get("stream", False))
|
|
if not messages or messages[-1].get("role") != "user":
|
|
raise HTTPException(
|
|
status.HTTP_400_BAD_REQUEST, "the last message must be a user turn"
|
|
)
|
|
|
|
_log.info(
|
|
"messages: actor=%s agent=%s stream=%s msgs=%d",
|
|
token_name,
|
|
agent.name,
|
|
stream_flag,
|
|
len(messages),
|
|
)
|
|
await audit.log(
|
|
runtime,
|
|
actor=f"token:{token_name}",
|
|
kind="messages",
|
|
agent_name=agent.name,
|
|
stream=stream_flag,
|
|
msgs=len(messages),
|
|
)
|
|
|
|
if isinstance(agent, ClaudeAgent) and runtime.conversations is not None:
|
|
conv, history = await _resolve(runtime.conversations, agent, messages)
|
|
turn_id = f"turn_{uuid.uuid4().hex[:12]}"
|
|
capture = TurnCapture()
|
|
events = runtime.conversations.turn(
|
|
conv,
|
|
messages=messages,
|
|
origin="user",
|
|
capture=capture,
|
|
turn_id=turn_id,
|
|
)
|
|
|
|
async def after(message: Message) -> None:
|
|
await _finish(
|
|
runtime,
|
|
conv,
|
|
messages=messages,
|
|
message=message,
|
|
capture=capture,
|
|
turn_id=turn_id,
|
|
history=history,
|
|
)
|
|
else:
|
|
options: dict[str, Any] = {}
|
|
if isinstance(body.get("temperature"), int | float):
|
|
options["temperature"] = body["temperature"]
|
|
events = backend.complete(
|
|
agent=agent, messages=messages, system=system_str, **options
|
|
)
|
|
|
|
async def after(message: Message) -> None:
|
|
await _broadcast_turn(
|
|
runtime,
|
|
agent_name=agent.name,
|
|
input_messages=messages,
|
|
output_message=message,
|
|
system=system_str,
|
|
)
|
|
|
|
if stream_flag:
|
|
return StreamingResponse(
|
|
_sse(events, model=model, after=after),
|
|
media_type="text/event-stream",
|
|
)
|
|
acc = StreamAccumulator()
|
|
async for ev in events:
|
|
acc.feed(ev)
|
|
message = acc.finalize(model=model)
|
|
await after(message)
|
|
return JSONResponse(content=message.model_dump(mode="json"))
|
|
|
|
return app
|
|
|
|
|
|
async def _resolve(
|
|
conversations: Conversations, agent: ClaudeAgent, messages: list[MessageParam]
|
|
) -> tuple[Conversation, list[dict[str, str]] | None]:
|
|
prior = [dict(m) for m in messages[:-1]]
|
|
if prior:
|
|
conv = await conversations.find_bound(
|
|
frontend=FRONTEND, external_id=fingerprint(prior)
|
|
)
|
|
if conv is not None:
|
|
if conv.agent_name != agent.name:
|
|
raise HTTPException(
|
|
status.HTTP_400_BAD_REQUEST,
|
|
f"this chat runs on {conv.agent_name!r}, not {agent.name!r}",
|
|
)
|
|
return conv, None
|
|
first = next(
|
|
(text_of(m.get("content")) for m in messages if m.get("role") == "user"), ""
|
|
)
|
|
try:
|
|
conv = await conversations.create(
|
|
kind="deep",
|
|
agent=agent.name,
|
|
origin=FRONTEND,
|
|
title=" ".join(first.split())[:_TITLE_LEN] or None,
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
|
|
await conversations.materialize(conv)
|
|
history = [
|
|
{"role": str(m.get("role", "")), "text": text_of(m.get("content"))}
|
|
for m in prior
|
|
]
|
|
return conv, history
|
|
|
|
|
|
async def _finish(
|
|
runtime: GatewayRuntime,
|
|
conv: Conversation,
|
|
*,
|
|
messages: list[MessageParam],
|
|
message: Message,
|
|
capture: TurnCapture,
|
|
turn_id: str,
|
|
history: list[dict[str, str]] | None,
|
|
) -> None:
|
|
reply = "\n\n".join(
|
|
getattr(b, "text", "")
|
|
for b in message.content
|
|
if getattr(b, "type", "") == "text"
|
|
).strip()
|
|
tail = capture.synthesized_messages or [
|
|
{"role": "assistant", "content": [{"type": "text", "text": reply}]}
|
|
]
|
|
await runtime.conversations.bind(
|
|
conv, frontend=FRONTEND, external_id=fingerprint([*messages, *tail])
|
|
)
|
|
runtime.bus.publish(
|
|
"reply",
|
|
conversation_id=conv.external_id,
|
|
turn_id=turn_id,
|
|
source=FRONTEND,
|
|
prompt=text_of(messages[-1].get("content")),
|
|
text=reply,
|
|
history=history,
|
|
)
|
|
|
|
|
|
async def _sse(
|
|
events: AsyncIterator[MessageStreamEvent],
|
|
*,
|
|
model: str,
|
|
after: Callable[[Message], Awaitable[None]],
|
|
) -> AsyncIterator[bytes]:
|
|
r"""Serialize an event stream to SSE, then hand the assembled ``Message`` on.
|
|
|
|
Each event becomes ``event: <type>\ndata: <json>\n\n`` - the shape
|
|
the Anthropic SDK's SSE decoder expects. Errors mid-stream are
|
|
swallowed into a synthetic ``error`` event so the client sees the
|
|
failure rather than a hung connection.
|
|
"""
|
|
acc = StreamAccumulator()
|
|
try:
|
|
async for ev in events:
|
|
acc.feed(ev)
|
|
payload = ev.model_dump_json()
|
|
yield f"event: {ev.type}\ndata: {payload}\n\n".encode()
|
|
except Exception as exc: # noqa: BLE001
|
|
_log.exception("backend stream failed")
|
|
err = json.dumps(
|
|
{"type": "error", "error": {"type": "api_error", "message": str(exc)}}
|
|
)
|
|
yield f"event: error\ndata: {err}\n\n".encode()
|
|
return
|
|
try:
|
|
await after(acc.finalize(model=model))
|
|
except Exception: # noqa: BLE001
|
|
_log.exception("post-turn hook failed")
|
|
|
|
|
|
async def _broadcast_turn(
|
|
runtime: GatewayRuntime,
|
|
*,
|
|
agent_name: str,
|
|
input_messages: list[MessageParam],
|
|
output_message: Message,
|
|
system: str | None,
|
|
) -> None:
|
|
if not runtime.turn_log_handlers:
|
|
return
|
|
record = TurnRecord(
|
|
agent_name=agent_name,
|
|
input_messages=list(input_messages),
|
|
output_message=output_message,
|
|
system=system,
|
|
source=FRONTEND,
|
|
)
|
|
for handler in runtime.turn_log_handlers:
|
|
try:
|
|
await handler(record)
|
|
except Exception: # noqa: BLE001
|
|
_log.exception("turn_log_handler raised; continuing")
|