File size: 3,711 Bytes
0703e2b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 | """MCP (Model Context Protocol) tools as FGAO workflow tools.
Works with any client exposing ``await call_tool(name, arguments)`` and
``await list_tools()``: ``mcp.Client`` (SDK 2.x) or ``mcp.ClientSession``
(SDK 1.x). FGAO decides *when* to call a tool from its declared effects; the
``inputs`` function builds the call's arguments from the workflow state and
``output`` turns the result into a state update.
from mcp import Client
from fgao import Workflow
from fgao.integrations.mcp import add_mcp_tool
async with Client("http://localhost:8000/mcp") as client:
wf = Workflow("ops")
wf.add_goal("ticket", lambda s: "ticket_id" in s)
add_mcp_tool(wf, client, "create_ticket", inputs=lambda s: {"title": s["title"]},
output=lambda s, r: {"ticket_id": r.data["id"]}, gain={"ticket": 0.9}, cost=1)
result = await wf.compile(budget=10).ainvoke({"title": "disk full"})
MCP calls are async, so these tools run through the async API
(``ainvoke`` / ``abatch`` / ``astream`` / ``Run.arun``).
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Mapping, Optional
from ..types import Action
from ..workflow import ToolResult, Workflow
@dataclass
class MCPResult:
"""Normalised tool result: concatenated text, structured content (if any), error flag."""
text: str
data: Any = None
is_error: bool = False
raw: Any = field(default=None, repr=False)
@classmethod
def of(cls, res: Any) -> "MCPResult":
content = getattr(res, "content", None) or []
text = "".join(getattr(b, "text", "") for b in content if getattr(b, "type", "text") == "text")
data = getattr(res, "structured_content", None)
if data is None:
data = getattr(res, "structuredContent", None)
err = getattr(res, "is_error", None)
if err is None:
err = getattr(res, "isError", False)
return cls(text=text, data=data, is_error=bool(err), raw=res)
async def list_mcp_tools(client: Any) -> List[Dict[str, Any]]:
"""[{"name", "description", "input_schema"}] for every tool the server offers."""
res = await client.list_tools()
out = []
for t in getattr(res, "tools", res):
schema = getattr(t, "input_schema", None) or getattr(t, "inputSchema", None)
out.append({"name": t.name, "description": getattr(t, "description", "") or "", "input_schema": schema})
return out
def add_mcp_tool(
wf: Workflow,
client: Any,
tool: str,
*,
inputs: Callable[..., Mapping[str, Any]],
output: Callable[[Any, MCPResult], Optional[Mapping[str, Any]]],
name: Optional[str] = None,
args: Optional[Mapping[str, Any]] = None,
timeout: Optional[float] = None,
**declared: Any,
) -> Action:
"""Register MCP tool ``tool`` as an FGAO tool.
inputs(state, **args) -> the tool arguments; output(state, MCPResult) -> state update.
An MCP error result is recorded as a failed call (no update), so FGAO's feedback learns from it.
``declared`` are the usual ``add_tool`` declarations (gain, harm, produces, requires, cost, group).
"""
async def afn(state: Any, **kw: Any):
call_kw = {"read_timeout_seconds": timeout} if timeout is not None else {}
res = MCPResult.of(await client.call_tool(tool, dict(inputs(state, **kw)), **call_kw))
if res.is_error:
return ToolResult(error=f"MCP tool {tool} failed: {res.text[:300]}")
return ToolResult(update=output(state, res))
return wf.add_tool(name or tool, afn, args=args, **declared)
__all__ = ["add_mcp_tool", "list_mcp_tools", "MCPResult"]
|