Coverage for src / osiris_cli / mcp.py: 0%

1""" 

2Minimal MCP (Model Context Protocol) client scaffold. 

3Refs for parity: Goose MCP (stdio/http/sse) and Crush MCP support. 

4Currently disabled by default; enable via settings.mcp_enabled and settings.mcp_servers. 

5""" 

6 

7import asyncio 

8import json 

9import os 

10from typing import Any, Dict 

11 

12import httpx 

13 

14from .config import settings 

15 

16 

17class MCPClient: 

18 def __init__(self): 

19 self.reload() 

20 

21 def reload(self): 

22 self.servers = settings.mcp_servers or {} 

23 

24 async def health(self) -> Dict[str, Any]: 

25 """Return basic health info for configured servers.""" 

26 results = {} 

27 for name, cfg in self.servers.items(): 

28 try: 

29 t = cfg.get("type") 

30 if t == "http": 

31 url = cfg.get("url") 

32 async with httpx.AsyncClient(timeout=10) as client: 

33 resp = await client.get(url) 

34 results[name] = {"status": "ok", "http_status": resp.status_code} 

35 elif t == "stdio": 

36 # try to run command --version or just spawn 

37 cmd = cfg.get("command") 

38 if cmd: 

39 proc = await asyncio.create_subprocess_exec( 

40 cmd, *cfg.get("args", []), 

41 stdin=asyncio.subprocess.PIPE, 

42 stdout=asyncio.subprocess.PIPE, 

43 stderr=asyncio.subprocess.PIPE 

44 ) 

45 proc.kill() 

46 results[name] = {"status": "ok", "note": "spawnable"} 

47 else: 

48 results[name] = {"status": "error", "error": "no command"} 

49 else: 

50 results[name] = {"status": "configured", "note": "health not implemented"} 

51 except Exception as e: 

52 results[name] = {"status": "error", "error": str(e)} 

53 return results 

54 

55 async def call(self, server: str, payload: Dict[str, Any]) -> Dict[str, Any]: 

56 if not settings.mcp_enabled: 

57 return {"error": "MCP disabled"} 

58 cfg = self.servers.get(server) 

59 if not cfg: 

60 return {"error": "Unknown MCP server"} 

61 t = cfg.get("type") 

62 if t == "http": 

63 return await self._call_http(cfg, payload) 

64 elif t == "stdio": 

65 return await self._call_stdio(cfg, payload) 

66 elif t == "sse": 

67 return {"error": "SSE not implemented"} 

68 return {"error": f"Unsupported MCP type: {t}"} 

69 

70 async def _call_http(self, cfg: dict, payload: Dict[str, Any]) -> Dict[str, Any]: 

71 url = cfg.get("url") 

72 headers_raw = cfg.get("headers", {}) 

73 # Expand environment placeholders like "${GITHUB_TOKEN}" in header values. 

74 headers: Dict[str, str] = {} 

75 for key, value in headers_raw.items(): 

76 if isinstance(value, str): 

77 headers[key] = os.path.expandvars(value) 

78 else: 

79 headers[key] = str(value) 

80 timeout = cfg.get("timeout", 20) 

81 method = (cfg.get("method") or "POST").upper() 

82 if not url: 

83 return {"error": "Missing url"} 

84 try: 

85 async with httpx.AsyncClient(timeout=timeout) as client: 

86 if method == "GET": 

87 resp = await client.get(url, params=payload, headers=headers) 

88 else: 

89 resp = await client.post(url, json=payload, headers=headers) 

90 return { 

91 "status": resp.status_code, 

92 "body": resp.text[:8000] 

93 } 

94 except Exception as e: 

95 return {"error": str(e)} 

96 

97 async def _call_stdio(self, cfg: dict, payload: Dict[str, Any]) -> Dict[str, Any]: 

98 cmd = cfg.get("command") 

99 args = cfg.get("args", []) 

100 env = cfg.get("env", {}) 

101 timeout = cfg.get("timeout", 20) 

102 if not cmd: 

103 return {"error": "No command provided"} 

104 try: 

105 proc = await asyncio.create_subprocess_exec( 

106 cmd, *args, 

107 stdin=asyncio.subprocess.PIPE, 

108 stdout=asyncio.subprocess.PIPE, 

109 stderr=asyncio.subprocess.PIPE, 

110 env={**env, **dict(**{k: str(v) for k, v in env.items()})} if env else None 

111 ) 

112 out, err = await asyncio.wait_for(proc.communicate(json.dumps(payload).encode()), timeout=timeout) 

113 return { 

114 "status": proc.returncode, 

115 "stdout": out.decode(errors="ignore")[:8000], 

116 "stderr": err.decode(errors="ignore")[:2000], 

117 } 

118 except Exception as e: 

119 return {"error": str(e)} 

120 

121 

122mcp_client = MCPClient()