diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7b482de..4d415c7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -51,5 +51,8 @@ jobs: SYNTHORA_AUTH_MODE: none SYNTHORA_SECRET_KEY: ci-smoke-secret-key-not-for-prod-32 SYNTHORA_CHECKPOINT_BACKEND: postgres + OPENAI_API_KEY: "" + OLLAMA_BASE_URL: "" + SYNTHORA_EMBEDDINGS: hash run: bash scripts/smoke.sh timeout-minutes: 45 diff --git a/apps/api/pyproject.toml b/apps/api/pyproject.toml index 8737b23..a63f0d6 100644 --- a/apps/api/pyproject.toml +++ b/apps/api/pyproject.toml @@ -18,6 +18,7 @@ dependencies = [ "pypdf>=5.0", "python-docx>=1.1", "mcp>=1.28.1", + "python-multipart>=0.0.32", ] [tool.uv.sources] diff --git a/apps/web/src/components/News.tsx b/apps/web/src/components/News.tsx index 9c79012..c1bd1a7 100644 --- a/apps/web/src/components/News.tsx +++ b/apps/web/src/components/News.tsx @@ -4,6 +4,7 @@ import { api, NewsItem, NewsSubscription } from "../api"; export function News() { const [subs, setSubs] = useState([]); const [items, setItems] = useState([]); + const [filterSubId, setFilterSubId] = useState(null); const [query, setQuery] = useState(""); const [cadence, setCadence] = useState("daily"); const [editingId, setEditingId] = useState(null); @@ -12,10 +13,12 @@ export function News() { const [busy, setBusy] = useState(false); const [error, setError] = useState(null); - async function refresh() { + async function refresh(subscriptionId?: string | null) { + const filter = + subscriptionId !== undefined ? subscriptionId : filterSubId; const [s, i] = await Promise.all([ api.listNewsSubscriptions(), - api.listNewsItems(), + api.listNewsItems(filter || undefined), ]); setSubs(s); setItems(i); @@ -25,6 +28,11 @@ export function News() { refresh().catch((e) => setError(String(e))); }, []); + function subLabel(subscriptionId: string): string { + const match = subs.find((s) => s.id === subscriptionId); + return match?.query || `${subscriptionId.slice(0, 8)}…`; + } + async function createSub() { if (!query.trim()) return; setBusy(true); @@ -59,7 +67,12 @@ export function News() { try { await api.deleteNewsSubscription(id); if (editingId === id) setEditingId(null); - await refresh(); + if (filterSubId === id) { + setFilterSubId(null); + await refresh(null); + } else { + await refresh(); + } } catch (e) { setError(String(e)); } finally { @@ -91,6 +104,18 @@ export function News() { } } + async function toggleFilter(sub: NewsSubscription) { + setError(null); + if (filterSubId === sub.id) { + setFilterSubId(null); + await refresh(null); + return; + } + setFilterSubId(sub.id); + await api.getNewsSubscription(sub.id); + await refresh(sub.id); + } + return (

News subscriptions

@@ -162,7 +187,14 @@ export function News() { ) : ( <>
- {s.query} + · {s.cadence}
@@ -198,10 +230,26 @@ export function News() {

Items

+ {filterSubId && ( +

+ Filtered to subscription: {subLabel(filterSubId)}{" "} + +

+ )} {items.length === 0 &&

No news items yet.

}
    {items.map((item) => (
  • + {subLabel(item.subscription_id)} · {item.title || item.url} diff --git a/docker-compose.yml b/docker-compose.yml index 8053e32..ed1566e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -81,6 +81,7 @@ services: OPENAI_BASE_URL: ${OPENAI_BASE_URL:-} TAVILY_API_KEY: ${TAVILY_API_KEY:-} OLLAMA_BASE_URL: ${OLLAMA_BASE_URL:-http://ollama:11434} + SYNTHORA_EMBEDDINGS: ${SYNTHORA_EMBEDDINGS:-} ports: - "${SYNTHORA_API_PORT:-8000}:8000" depends_on: diff --git a/docs/feature-parity.md b/docs/feature-parity.md index 63d6614..7dfdb57 100644 --- a/docs/feature-parity.md +++ b/docs/feature-parity.md @@ -61,7 +61,7 @@ See also [parity-audit.md](parity-audit.md). | LLM provider abstraction + think-tag handling | ✅ | 11 providers | | Research history + export md/html/pdf | ✅ | API + web buttons; structured HTML/PDF export | | Docker Compose self-host | ✅ | `docker-compose.yml` | -| Python SDK | ✅ | `packages/sdk` — full REST mirror incl. upload, export download, news GET | +| Python SDK | ✅ | `packages/sdk` — sync + async clients, WebSocket events, full REST mirror | | Document library + RAG (`collection` engine) | ✅ | documents API + `document_index` | | Provider settings persistence | ✅ | `/api/v1/settings` + Settings UI; resolvers prefer workspace overlay then env; GET responses redact secrets | | MCP server exposing Synthora tools | ✅ | `/api/v1/mcp/tools/*` REST + `/mcp` streamable HTTP; optional ``config`` on ``start_research`` | @@ -130,6 +130,11 @@ code, blockquotes, ordered lists, tables, rules); PDF via fpdf2 ``write_html`` (preserves structure); web ``health``/``ready``/MCP REST wrappers; History session drill-down via ``getSession``. +Closed on ``feat/mcp-async-hardening``: outbound MCP HTTP fallback fail-loud + +integration tests; ``AsyncSynthoraClient`` with ``iter_run_events`` WebSocket; +News subscription filter + ``getNewsSubscription`` UX; smoke validates export +formats and document upload. + No known functional gaps remain beyond explicit non-goals below. Chat remains session-scoped ``fast_research`` with prior-report memory — diff --git a/packages/adapters/src/synthora/adapters/embeddings.py b/packages/adapters/src/synthora/adapters/embeddings.py index 5bde8e8..bb79cbd 100644 --- a/packages/adapters/src/synthora/adapters/embeddings.py +++ b/packages/adapters/src/synthora/adapters/embeddings.py @@ -183,6 +183,9 @@ def resolve(self, model_id: str) -> EmbeddingModel: def resolve_default_embeddings() -> EmbeddingModel: """Prefer OpenAI when keyed, else Ollama when base URL set, else hash.""" + forced = _env("SYNTHORA_EMBEDDINGS", default="").strip().lower() + if forced in ("hash", "offline"): + return HashEmbeddings() if _env("OPENAI_API_KEY"): return OpenAIEmbeddings() if _env("OLLAMA_BASE_URL") or _env("OLLAMA_EMBED_MODEL"): diff --git a/packages/adapters/src/synthora/adapters/mcp_client.py b/packages/adapters/src/synthora/adapters/mcp_client.py index 9e4e789..383bf39 100644 --- a/packages/adapters/src/synthora/adapters/mcp_client.py +++ b/packages/adapters/src/synthora/adapters/mcp_client.py @@ -229,6 +229,8 @@ async def _http_tools_list( tools = data.get("tools") if isinstance(data, dict) else None if isinstance(tools, list): return tools + if resp.status_code not in (404, 405): + resp.raise_for_status() # Minimal JSON-RPC resp = await client.post( base_url, @@ -243,12 +245,16 @@ async def _http_tools_list( if resp.status_code >= 400: resp.raise_for_status() data = resp.json() + if isinstance(data, dict) and "error" in data: + err = data["error"] + message = err.get("message") if isinstance(err, dict) else str(err) + raise RuntimeError(f"MCP tools/list failed: {message}") result = data.get("result") if isinstance(data, dict) else None if isinstance(result, dict) and isinstance(result.get("tools"), list): return result["tools"] if isinstance(result, list): return result - return [] + raise RuntimeError("MCP tools/list returned no tools") async def _http_tools_call( @@ -273,6 +279,8 @@ async def _http_tools_call( if "result" in data: return str(data["result"]) return str(data) + if resp.status_code not in (404, 405): + resp.raise_for_status() resp = await client.post( base_url, json={ @@ -286,8 +294,17 @@ async def _http_tools_call( resp.raise_for_status() data = resp.json() if isinstance(data, dict): - if "result" in data: - return str(data["result"]) if "error" in data: - return f"error: {data['error']}" + err = data["error"] + message = err.get("message") if isinstance(err, dict) else str(err) + raise RuntimeError(f"MCP tools/call failed: {message}") + if "result" in data: + result = data["result"] + if isinstance(result, dict) and "content" in result: + blocks = result["content"] + if isinstance(blocks, list): + for block in blocks: + if isinstance(block, dict) and block.get("type") == "text": + return str(block.get("text", "")) + return str(result) return str(data) diff --git a/packages/persistence/src/synthora/persistence/repositories.py b/packages/persistence/src/synthora/persistence/repositories.py index 1232c0a..712ce23 100644 --- a/packages/persistence/src/synthora/persistence/repositories.py +++ b/packages/persistence/src/synthora/persistence/repositories.py @@ -596,6 +596,7 @@ async def create( created_at=document.created_at, ) ) + await s.flush() for chunk in chunks or []: s.add( DocumentChunkRow( diff --git a/packages/sdk/pyproject.toml b/packages/sdk/pyproject.toml index b3715b2..fb17a8f 100644 --- a/packages/sdk/pyproject.toml +++ b/packages/sdk/pyproject.toml @@ -3,7 +3,7 @@ name = "synthora-sdk" version = "0.1.0" description = "Synthora Python client SDK" requires-python = ">=3.11" -dependencies = ["httpx>=0.27"] +dependencies = ["httpx>=0.27", "websockets>=12.0"] [build-system] requires = ["hatchling"] diff --git a/packages/sdk/src/synthora/sdk/__init__.py b/packages/sdk/src/synthora/sdk/__init__.py index c522591..de2f4d3 100644 --- a/packages/sdk/src/synthora/sdk/__init__.py +++ b/packages/sdk/src/synthora/sdk/__init__.py @@ -1,5 +1,6 @@ """Synthora Python SDK (R-LDR-8).""" +from synthora.sdk.async_client import AsyncSynthoraClient from synthora.sdk.client import SynthoraClient -__all__ = ["SynthoraClient"] +__all__ = ["AsyncSynthoraClient", "SynthoraClient"] diff --git a/packages/sdk/src/synthora/sdk/async_client.py b/packages/sdk/src/synthora/sdk/async_client.py new file mode 100644 index 0000000..3044d9f --- /dev/null +++ b/packages/sdk/src/synthora/sdk/async_client.py @@ -0,0 +1,348 @@ +"""Async Python client for the Synthora API.""" + +from __future__ import annotations + +import asyncio +import json +import time +from collections.abc import AsyncIterator +from typing import Any, Optional + +import httpx + + +class AsyncSynthoraClient: + """Async client mirroring :class:`SynthoraClient`.""" + + def __init__( + self, + base_url: str = "http://localhost:8000", + *, + token: Optional[str] = None, + timeout: float = 30.0, + ) -> None: + self.base_url = base_url.rstrip("/") + self.token = token + self._client = httpx.AsyncClient(base_url=self.base_url, timeout=timeout) + + async def register(self, username: str, password: str) -> str: + data = await self._post( + "/api/v1/auth/register", {"username": username, "password": password} + ) + self.token = data["token"] + return self.token + + async def login(self, username: str, password: str) -> str: + data = await self._post( + "/api/v1/auth/login", {"username": username, "password": password} + ) + self.token = data["token"] + return self.token + + async def create_session( + self, title: str = "Untitled research", tags: Optional[list[str]] = None + ) -> dict: + return await self._post("/api/v1/sessions", {"title": title, "tags": tags or []}) + + async def list_sessions(self) -> list[dict]: + return (await self._get("/api/v1/sessions"))["sessions"] + + async def get_session(self, session_id: str) -> dict: + return await self._get(f"/api/v1/sessions/{session_id}") + + async def delete_session(self, session_id: str) -> dict: + return await self._delete(f"/api/v1/sessions/{session_id}") + + async def start_research( + self, + question: str, + *, + pipeline_id: str = "deep_research", + session_id: Optional[str] = None, + config: Optional[dict[str, Any]] = None, + ) -> str: + body: dict[str, Any] = { + "question": question, + "pipeline_id": pipeline_id, + "config": config, + } + if session_id: + body["session_id"] = session_id + data = await self._post("/api/v1/research", body) + return data["run_id"] + + async def get_run(self, run_id: str) -> dict: + return await self._get(f"/api/v1/research/{run_id}") + + async def list_runs(self, *, session_id: Optional[str] = None) -> list[dict]: + path = "/api/v1/research" + if session_id: + path = f"{path}?session_id={session_id}" + return (await self._get(path))["runs"] + + async def delete_run(self, run_id: str) -> dict: + return await self._delete(f"/api/v1/research/{run_id}") + + async def clear_history(self) -> dict: + return await self._post("/api/v1/research/clear", {}) + + async def cancel(self, run_id: str) -> dict: + return await self._post(f"/api/v1/research/{run_id}/cancel", {}) + + async def resume(self, run_id: str, answer: str) -> dict: + return await self._post( + f"/api/v1/research/{run_id}/resume", {"answer": answer} + ) + + async def steer(self, run_id: str, message: str) -> dict: + return await self._post( + f"/api/v1/research/{run_id}/steer", {"message": message} + ) + + async def get_report(self, run_id: str) -> dict: + return await self._get(f"/api/v1/research/{run_id}/report") + + async def get_events(self, run_id: str) -> list[dict]: + return (await self._get(f"/api/v1/research/{run_id}/events"))["events"] + + async def get_knowledge_map(self, run_id: str) -> dict: + return await self._get(f"/api/v1/research/{run_id}/knowledge-map") + + async def get_discourse(self, run_id: str) -> list[dict]: + return (await self._get(f"/api/v1/research/{run_id}/discourse"))["turns"] + + def export_url(self, run_id: str, fmt: str = "markdown") -> str: + return f"{self.base_url}/api/v1/research/{run_id}/export?format={fmt}" + + def events_ws_url(self, run_id: str) -> str: + ws_base = self.base_url.replace("https://", "wss://").replace( + "http://", "ws://" + ) + suffix = f"?token={self.token}" if self.token else "" + return f"{ws_base}/api/v1/research/{run_id}/events/ws{suffix}" + + async def download_export(self, run_id: str, fmt: str = "markdown") -> bytes: + resp = await self._client.get( + f"/api/v1/research/{run_id}/export", + params={"format": fmt}, + headers=self._headers(), + ) + resp.raise_for_status() + return resp.content + + async def list_pipelines(self) -> list[dict]: + return (await self._get("/api/v1/pipelines"))["pipelines"] + + async def list_providers(self) -> dict: + return await self._get("/api/v1/providers") + + async def wait_for_report( + self, run_id: str, *, poll_seconds: float = 2.0, timeout: float = 1800.0 + ) -> dict: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + run = await self.get_run(run_id) + status = run["status"] + if status == "completed": + return await self.get_report(run_id) + if status == "awaiting_input": + raise RuntimeError( + f"run {run_id} is awaiting_input; call resume() with an answer" + ) + if status in ("failed", "cancelled"): + raise RuntimeError( + f"run {run_id} {status}: {run.get('error')}" + ) + await asyncio.sleep(poll_seconds) + raise TimeoutError(f"run {run_id} did not finish within {timeout}s") + + async def iter_run_events(self, run_id: str) -> AsyncIterator[dict]: + """Stream live events from the run WebSocket.""" + import websockets + + url = self.events_ws_url(run_id) + headers = [] + if self.token: + headers.append(("Authorization", f"Bearer {self.token}")) + async with websockets.connect(url, additional_headers=headers) as ws: + while True: + raw = await ws.recv() + yield json.loads(raw) + + async def chat(self, message: str, *, session_id: Optional[str] = None) -> dict: + body: dict[str, Any] = {"message": message} + if session_id: + body["session_id"] = session_id + return await self._post("/api/v1/chat", body) + + async def followup( + self, run_id: str, question: str, *, pipeline_id: Optional[str] = None + ) -> dict: + body: dict[str, Any] = {"question": question} + if pipeline_id: + body["pipeline_id"] = pipeline_id + return await self._post(f"/api/v1/research/{run_id}/followup", body) + + async def list_documents(self) -> list[dict]: + return (await self._get("/api/v1/documents"))["documents"] + + async def create_document( + self, title: str, content: str, *, url: Optional[str] = None + ) -> dict: + body: dict[str, Any] = {"title": title, "content": content} + if url: + body["url"] = url + return await self._post("/api/v1/documents", body) + + async def upload_document( + self, + file: str | bytes, + *, + filename: str = "upload.txt", + title: Optional[str] = None, + ) -> dict: + if isinstance(file, str): + with open(file, "rb") as handle: + payload = handle.read() + filename = filename or file.rsplit("/", 1)[-1] + else: + payload = file + files = {"file": (filename, payload, "application/octet-stream")} + data: dict[str, str] = {} + if title: + data["title"] = title + resp = await self._client.post( + "/api/v1/documents/upload", + files=files, + data=data or None, + headers=self._headers(), + ) + resp.raise_for_status() + return resp.json() + + async def delete_document(self, document_id: str) -> dict: + return await self._delete(f"/api/v1/documents/{document_id}") + + async def search_documents( + self, query: str, *, max_results: int = 5 + ) -> list[dict]: + return ( + await self._post( + "/api/v1/documents/search", + {"query": query, "max_results": max_results}, + ) + ).get("results", []) + + async def list_news_subscriptions(self) -> list[dict]: + return (await self._get("/api/v1/news/subscriptions"))["subscriptions"] + + async def get_news_subscription(self, subscription_id: str) -> dict: + return await self._get(f"/api/v1/news/subscriptions/{subscription_id}") + + async def create_news_subscription( + self, query: str, *, cadence: str = "daily" + ) -> dict: + return await self._post( + "/api/v1/news/subscriptions", {"query": query, "cadence": cadence} + ) + + async def delete_news_subscription(self, subscription_id: str) -> dict: + return await self._delete(f"/api/v1/news/subscriptions/{subscription_id}") + + async def fetch_news_subscription(self, subscription_id: str) -> dict: + return await self._post( + f"/api/v1/news/subscriptions/{subscription_id}/fetch", {} + ) + + async def update_news_subscription( + self, + subscription_id: str, + *, + query: Optional[str] = None, + cadence: Optional[str] = None, + ) -> dict: + body: dict[str, Any] = {} + if query is not None: + body["query"] = query + if cadence is not None: + body["cadence"] = cadence + return await self._patch( + f"/api/v1/news/subscriptions/{subscription_id}", body + ) + + async def list_news_items( + self, *, subscription_id: Optional[str] = None + ) -> list[dict]: + path = "/api/v1/news/items" + if subscription_id: + path = f"{path}?subscription_id={subscription_id}" + return (await self._get(path))["items"] + + async def list_settings(self) -> list[dict]: + return (await self._get("/api/v1/settings"))["settings"] + + async def get_setting(self, key: str) -> dict: + return await self._get(f"/api/v1/settings/{key}") + + async def put_setting(self, key: str, value: dict[str, Any]) -> dict: + return await self._put(f"/api/v1/settings/{key}", {"value": value}) + + async def get_run_metrics(self, run_id: str) -> dict: + return await self._get(f"/api/v1/research/{run_id}/metrics") + + async def metrics_summary(self) -> dict: + return await self._get("/api/v1/metrics/summary") + + async def mcp_tools_list(self) -> dict: + return await self._post("/api/v1/mcp/tools/list", {}) + + async def mcp_tools_call( + self, name: str, arguments: Optional[dict] = None + ) -> dict: + return await self._post( + "/api/v1/mcp/tools/call", + {"name": name, "arguments": arguments or {}}, + ) + + async def health(self) -> dict: + return await self._get("/health") + + async def ready(self) -> dict: + return await self._get("/ready") + + def _headers(self) -> dict: + return {"Authorization": f"Bearer {self.token}"} if self.token else {} + + async def _get(self, path: str) -> dict: + resp = await self._client.get(path, headers=self._headers()) + resp.raise_for_status() + return resp.json() + + async def _post(self, path: str, body: dict) -> dict: + resp = await self._client.post(path, json=body, headers=self._headers()) + resp.raise_for_status() + return resp.json() + + async def _put(self, path: str, body: dict) -> dict: + resp = await self._client.put(path, json=body, headers=self._headers()) + resp.raise_for_status() + return resp.json() + + async def _patch(self, path: str, body: dict) -> dict: + resp = await self._client.patch(path, json=body, headers=self._headers()) + resp.raise_for_status() + return resp.json() + + async def _delete(self, path: str) -> dict: + resp = await self._client.delete(path, headers=self._headers()) + resp.raise_for_status() + return resp.json() + + async def aclose(self) -> None: + await self._client.aclose() + + async def __aenter__(self) -> AsyncSynthoraClient: + return self + + async def __aexit__(self, *args) -> None: + await self.aclose() diff --git a/packages/sdk/src/synthora/sdk/client.py b/packages/sdk/src/synthora/sdk/client.py index 98403e5..68efff6 100644 --- a/packages/sdk/src/synthora/sdk/client.py +++ b/packages/sdk/src/synthora/sdk/client.py @@ -113,6 +113,14 @@ def get_discourse(self, run_id: str) -> list[dict]: def export_url(self, run_id: str, fmt: str = "markdown") -> str: return f"{self.base_url}/api/v1/research/{run_id}/export?format={fmt}" + def events_ws_url(self, run_id: str) -> str: + """WebSocket URL for live run events (append ``?token=`` in session mode).""" + ws_base = self.base_url.replace("https://", "wss://").replace( + "http://", "ws://" + ) + suffix = f"?token={self.token}" if self.token else "" + return f"{ws_base}/api/v1/research/{run_id}/events/ws{suffix}" + def download_export(self, run_id: str, fmt: str = "markdown") -> bytes: """Download export bytes with auth (session mode safe).""" resp = self._client.get( diff --git a/scripts/smoke.sh b/scripts/smoke.sh index 8414932..1f86b19 100755 --- a/scripts/smoke.sh +++ b/scripts/smoke.sh @@ -4,6 +4,12 @@ set -euo pipefail cd "$(dirname "$0")/.." +# Compose defaults OLLAMA_BASE_URL to the optional ollama profile service; smoke +# uses deterministic hash embeddings unless callers override these explicitly. +export OPENAI_API_KEY="${OPENAI_API_KEY:-}" +export OLLAMA_BASE_URL="${OLLAMA_BASE_URL:-}" +export SYNTHORA_EMBEDDINGS="${SYNTHORA_EMBEDDINGS:-hash}" + echo "==> validating compose file" docker compose config --quiet @@ -104,4 +110,26 @@ done REPORT=$(curl -fsS "http://localhost:${SYNTHORA_API_PORT:-8000}/api/v1/research/${RUN_ID}/report") echo "$REPORT" | python3 -c 'import json,sys; d=json.load(sys.stdin); assert d.get("report_markdown"), d; print("report ok:", d["report_markdown"][:80])' +echo "==> export formats" +curl -fsS "http://localhost:${SYNTHORA_API_PORT:-8000}/api/v1/research/${RUN_ID}/export?format=markdown" | head -c 120 +echo +curl -fsS "http://localhost:${SYNTHORA_API_PORT:-8000}/api/v1/research/${RUN_ID}/export?format=html" | head -c 120 +echo +PDF_HEAD=$(curl -fsS "http://localhost:${SYNTHORA_API_PORT:-8000}/api/v1/research/${RUN_ID}/export?format=pdf" | head -c 4) +[[ "$PDF_HEAD" == "%PDF" ]] || { echo "pdf export failed: $PDF_HEAD"; exit 1; } +echo "pdf export ok" + +echo "==> document upload" +printf 'smoke document library content' >/tmp/synthora-smoke-doc.txt +UPLOAD_CODE=$(curl -sS -o /tmp/synthora-upload.json -w "%{http_code}" -X POST "http://localhost:${SYNTHORA_API_PORT:-8000}/api/v1/documents/upload" -F "file=@/tmp/synthora-smoke-doc.txt;filename=smoke.txt" -F "title=Smoke doc") +if [[ "$UPLOAD_CODE" != "201" ]]; then + echo "document upload failed: HTTP $UPLOAD_CODE" + cat /tmp/synthora-upload.json || true + echo + docker compose logs --tail=60 api || true + exit 1 +fi +UPLOAD=$(cat /tmp/synthora-upload.json) +echo "$UPLOAD" | python3 -c 'import json,sys; d=json.load(sys.stdin); assert d.get("id"), d; print("upload ok:", d["id"])' + echo "smoke test passed" diff --git a/tests/test_mcp_client.py b/tests/test_mcp_client.py new file mode 100644 index 0000000..0ec661a --- /dev/null +++ b/tests/test_mcp_client.py @@ -0,0 +1,99 @@ +"""Outbound MCP client HTTP fallback tests.""" + +from __future__ import annotations + +import json + +import httpx +import pytest +from httpx import ASGITransport +from synthora.adapters.mcp_client import ( + _http_tools_call, + _http_tools_list, + load_mcp_tools, +) + +from tests.test_platform import fake_run_config + +pytest_plugins = ("tests.test_platform",) + + +@pytest.fixture +def route_mcp_http(platform, monkeypatch): + _client, app = platform + + class _RoutingAsyncClient(httpx.AsyncClient): + def __init__(self, *args, **kwargs): + kwargs["transport"] = ASGITransport(app=app) + kwargs["base_url"] = "http://127.0.0.1:8000" + super().__init__(*args, **kwargs) + + monkeypatch.setattr( + "synthora.adapters.mcp_client.httpx.AsyncClient", _RoutingAsyncClient + ) + + +@pytest.mark.asyncio +async def test_http_tools_list_via_rest(route_mcp_http): + tools = await _http_tools_list("http://127.0.0.1:8000", headers={}) + names = {t["name"] for t in tools} + assert names == { + "start_research", + "get_run_status", + "get_report", + "search_documents", + } + + +@pytest.mark.asyncio +async def test_http_tools_call_via_rest(route_mcp_http): + content = await _http_tools_call( + "http://127.0.0.1:8000", + "start_research", + { + "question": "MCP outbound test", + "pipeline_id": "fast_research", + "config": fake_run_config(), + }, + ) + payload = json.loads(content) + assert payload["run_id"] + assert payload["status"] == "queued" + + +@pytest.mark.asyncio +async def test_http_tools_call_raises_on_jsonrpc_error(monkeypatch): + async def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/api/v1/mcp/tools/call"): + return httpx.Response(404) + return httpx.Response( + 200, + json={"jsonrpc": "2.0", "id": 1, "error": {"message": "denied"}}, + ) + + transport = httpx.MockTransport(handler) + + class _MockAsyncClient(httpx.AsyncClient): + def __init__(self, *args, **kwargs): + kwargs["transport"] = transport + kwargs["base_url"] = "http://evil.test" + super().__init__(*args, **kwargs) + + monkeypatch.setattr( + "synthora.adapters.mcp_client.httpx.AsyncClient", _MockAsyncClient + ) + with pytest.raises(RuntimeError, match="tools/call failed"): + await _http_tools_call("http://evil.test", "x", {}) + + +@pytest.mark.asyncio +async def test_load_mcp_tools_http_fallback(route_mcp_http): + tools = await load_mcp_tools( + {"servers": [{"url": "http://127.0.0.1:8000", "transport": "http"}]} + ) + assert {t.name for t in tools} == { + "start_research", + "get_run_status", + "get_report", + "search_documents", + } diff --git a/tests/test_sdk.py b/tests/test_sdk.py index 534b431..943822e 100644 --- a/tests/test_sdk.py +++ b/tests/test_sdk.py @@ -5,6 +5,7 @@ import json from typing import Any, Optional +import httpx import pytest from synthora.sdk.client import SynthoraClient @@ -169,3 +170,23 @@ def test_sdk_download_export(platform, sdk): def test_sdk_health_and_ready(sdk): assert sdk.health()["status"] == "ok" assert sdk.ready()["status"] == "ready" + + +@pytest.mark.asyncio +async def test_async_sdk_health_and_mcp(platform): + from httpx import ASGITransport + from synthora.sdk.async_client import AsyncSynthoraClient + + _client, app = platform + async with AsyncSynthoraClient("http://testserver") as sdk: + sdk._client = httpx.AsyncClient( + transport=ASGITransport(app=app), base_url="http://testserver" + ) + assert (await sdk.health())["status"] == "ok" + tools = await sdk.mcp_tools_list() + assert {t["name"] for t in tools["tools"]} == { + "start_research", + "get_run_status", + "get_report", + "search_documents", + } diff --git a/uv.lock b/uv.lock index db99702..12cf920 100644 --- a/uv.lock +++ b/uv.lock @@ -2098,6 +2098,7 @@ dependencies = [ { name = "pyjwt" }, { name = "pypdf" }, { name = "python-docx" }, + { name = "python-multipart" }, { name = "redis" }, { name = "synthora-adapters" }, { name = "synthora-core" }, @@ -2116,6 +2117,7 @@ requires-dist = [ { name = "pyjwt", specifier = ">=2.9" }, { name = "pypdf", specifier = ">=5.0" }, { name = "python-docx", specifier = ">=1.1" }, + { name = "python-multipart", specifier = ">=0.0.32" }, { name = "redis", specifier = ">=5.0" }, { name = "synthora-adapters", editable = "packages/adapters" }, { name = "synthora-core", editable = "packages/core" }, @@ -2205,10 +2207,14 @@ version = "0.1.0" source = { editable = "packages/sdk" } dependencies = [ { name = "httpx" }, + { name = "websockets" }, ] [package.metadata] -requires-dist = [{ name = "httpx", specifier = ">=0.27" }] +requires-dist = [ + { name = "httpx", specifier = ">=0.27" }, + { name = "websockets", specifier = ">=12.0" }, +] [[package]] name = "synthora-worker"