bvandeusen
a7160772bf
- Parse multi-line SSE format correctly (event: on separate line, chunk in data.chunk) - Retry stream GET up to 10x with 300ms backoff when 404 (buffer not ready yet) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
96 lines
3.1 KiB
Python
96 lines
3.1 KiB
Python
"""MCP tools for Fable chat — create conversations and stream responses."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from typing import Any
|
|
|
|
from fable_mcp.client import FableClient, FableAPIError
|
|
|
|
|
|
async def list_conversations(
|
|
client: FableClient,
|
|
*,
|
|
limit: int = 20,
|
|
offset: int = 0,
|
|
) -> dict[str, Any]:
|
|
"""List MCP chat conversations."""
|
|
params: dict[str, Any] = {"limit": limit, "offset": offset, "type": "mcp"}
|
|
return await client.get("/api/chat/conversations", params=params)
|
|
|
|
|
|
async def send_message(
|
|
client: FableClient,
|
|
*,
|
|
message: str,
|
|
conversation_id: str | None = None,
|
|
think: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Send a message to Fable and return the full assistant response.
|
|
|
|
Creates a new MCP conversation if conversation_id is None.
|
|
Posts the user message to start generation, then streams the SSE buffer.
|
|
|
|
SSE event format:
|
|
id: <N>
|
|
event: <type> # chunk | done | tool_call | status | ...
|
|
data: <json>
|
|
|
|
Returns:
|
|
Dict with:
|
|
- "conversation_id": str
|
|
- "response": str (full assistant message)
|
|
- "tool_calls": list of any tool_call events observed
|
|
"""
|
|
if conversation_id is None:
|
|
conv = await client.post(
|
|
"/api/chat/conversations",
|
|
json={"conversation_type": "mcp"},
|
|
)
|
|
conversation_id = str(conv["id"])
|
|
|
|
# Start generation
|
|
await client.post(
|
|
f"/api/chat/conversations/{conversation_id}/messages",
|
|
json={"content": message, "think": think},
|
|
)
|
|
|
|
tokens: list[str] = []
|
|
tool_calls: list[Any] = []
|
|
|
|
stream_path = f"/api/chat/conversations/{conversation_id}/generation/stream"
|
|
|
|
# Retry connecting to the stream briefly — the background task may not have
|
|
# created the generation buffer by the time we issue the GET.
|
|
for attempt in range(10):
|
|
try:
|
|
event_type: str | None = None
|
|
async for raw_line in client.stream_get(stream_path):
|
|
if raw_line.startswith("event: "):
|
|
event_type = raw_line[len("event: "):].strip()
|
|
elif raw_line.startswith("data: "):
|
|
payload_str = raw_line[len("data: "):]
|
|
try:
|
|
data = json.loads(payload_str)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if event_type == "chunk":
|
|
tokens.append(data.get("chunk", ""))
|
|
elif event_type == "tool_call":
|
|
tool_calls.append(data.get("tool_call", data))
|
|
elif event_type == "done":
|
|
break
|
|
event_type = None # reset after consuming data line
|
|
break # stream completed successfully
|
|
except FableAPIError as exc:
|
|
if exc.status_code == 404 and attempt < 9:
|
|
await asyncio.sleep(0.3)
|
|
continue
|
|
raise
|
|
|
|
return {
|
|
"conversation_id": conversation_id,
|
|
"response": "".join(tokens),
|
|
"tool_calls": tool_calls,
|
|
}
|