fix: MCP client SSE transport — add correct Accept headers and SSE response parsing
All checks were successful
Build and Deploy DevOpsDash / build-image (push) Successful in 6s
All checks were successful
Build and Deploy DevOpsDash / build-image (push) Successful in 6s
FastMCP 2.0 streamable HTTP requires Accept: application/json, text/event-stream
and can return text/event-stream responses with data: {...} SSE lines.
This commit is contained in:
@@ -1,7 +1,14 @@
|
|||||||
"""MCP proxy client — calls DevOpsMCP's worklog/standup tools over HTTP MCP protocol."""
|
"""MCP proxy client — calls DevOpsMCP's worklog/standup tools over HTTP MCP protocol.
|
||||||
|
|
||||||
|
FastMCP 2.0 uses streamable HTTP transport (MCP spec §6.3.3):
|
||||||
|
- POST to /mcp with Accept: application/json, text/event-stream
|
||||||
|
- Server may respond with JSON or SSE stream (text/event-stream)
|
||||||
|
- SSE responses have `data: {...}` lines containing JSON-RPC messages
|
||||||
|
"""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
import os
|
import os
|
||||||
from typing import Any, Dict, Optional
|
from typing import Any, Dict, Optional
|
||||||
|
|
||||||
@@ -10,25 +17,36 @@ import httpx
|
|||||||
MCP_URL = os.environ.get("DEVOPS_MCP_URL", "http://localhost:8000")
|
MCP_URL = os.environ.get("DEVOPS_MCP_URL", "http://localhost:8000")
|
||||||
_MCP_ENDPOINT = f"{MCP_URL}/mcp"
|
_MCP_ENDPOINT = f"{MCP_URL}/mcp"
|
||||||
|
|
||||||
|
_HEADERS = {
|
||||||
|
"Content-Type": "application/json",
|
||||||
|
"Accept": "application/json, text/event-stream",
|
||||||
|
}
|
||||||
|
|
||||||
async def _call_tool(tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
|
|
||||||
"""Call a DevOpsMCP tool via MCP JSON-RPC 2.0 over HTTP."""
|
def _parse_sse_data(text: str) -> Dict[str, Any]:
|
||||||
payload = {
|
"""Extract JSON-RPC result from SSE stream data lines."""
|
||||||
"jsonrpc": "2.0",
|
for line in text.splitlines():
|
||||||
"method": "tools/call",
|
line = line.strip()
|
||||||
"params": {"name": tool_name, "arguments": arguments},
|
if line.startswith("data:"):
|
||||||
"id": 1,
|
chunk = line[5:].strip()
|
||||||
}
|
if not chunk or chunk == "[DONE]":
|
||||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
continue
|
||||||
resp = await client.post(_MCP_ENDPOINT, json=payload)
|
try:
|
||||||
resp.raise_for_status()
|
msg = json.loads(chunk)
|
||||||
data = resp.json()
|
if "result" in msg or "error" in msg:
|
||||||
|
return msg
|
||||||
|
except json.JSONDecodeError:
|
||||||
|
continue
|
||||||
|
return {}
|
||||||
|
|
||||||
|
|
||||||
|
def _extract_result(data: Dict[str, Any]) -> Dict[str, Any]:
|
||||||
|
"""Extract tool result content from JSON-RPC response."""
|
||||||
if "error" in data:
|
if "error" in data:
|
||||||
raise RuntimeError(f"MCP error: {data['error']}")
|
raise RuntimeError(f"MCP error: {data['error']}")
|
||||||
result = data.get("result", {})
|
result = data.get("result", {})
|
||||||
content = result.get("content", [])
|
content = result.get("content", [])
|
||||||
if content and isinstance(content[0], dict):
|
if content and isinstance(content[0], dict):
|
||||||
import json
|
|
||||||
text = content[0].get("text", "{}")
|
text = content[0].get("text", "{}")
|
||||||
try:
|
try:
|
||||||
return json.loads(text)
|
return json.loads(text)
|
||||||
@@ -37,6 +55,27 @@ async def _call_tool(tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any
|
|||||||
return result
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
async def _call_tool(tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
|
||||||
|
"""Call a DevOpsMCP tool via MCP JSON-RPC 2.0 streamable HTTP transport."""
|
||||||
|
payload = {
|
||||||
|
"jsonrpc": "2.0",
|
||||||
|
"method": "tools/call",
|
||||||
|
"params": {"name": tool_name, "arguments": arguments},
|
||||||
|
"id": 1,
|
||||||
|
}
|
||||||
|
async with httpx.AsyncClient(timeout=60.0) as client:
|
||||||
|
resp = await client.post(_MCP_ENDPOINT, json=payload, headers=_HEADERS)
|
||||||
|
resp.raise_for_status()
|
||||||
|
|
||||||
|
content_type = resp.headers.get("content-type", "")
|
||||||
|
if "text/event-stream" in content_type:
|
||||||
|
data = _parse_sse_data(resp.text)
|
||||||
|
else:
|
||||||
|
data = resp.json()
|
||||||
|
|
||||||
|
return _extract_result(data)
|
||||||
|
|
||||||
|
|
||||||
async def get_worklog(
|
async def get_worklog(
|
||||||
context: str = "egmont",
|
context: str = "egmont",
|
||||||
days: int = 7,
|
days: int = 7,
|
||||||
|
|||||||
Reference in New Issue
Block a user