apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-21 19:21:03 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Make mcp-tool using configuration service for information about where the

Browse source 
MCP services are.
This commit is contained in:
Cyber MacGeddon 2025-07-07 23:40:46 +01:00
parent 9790a994ae
commit 1f1e7b1940
1 changed files with 36 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

45
trustgraph-flow/trustgraph/agent/mcp_tool/service.py
View file

@ -4,11 +4,12 @@ MCP tool-calling service, calls an external MCP tool. Input is
name + parameters, output is the response, either a string or an object. name + parameters, output is the response, either a string or an object.
""" """
from ... base import ToolService import json
from mcp.client.streamable_http import streamablehttp_client from mcp.client.streamable_http import streamablehttp_client
from mcp import ClientSession from mcp import ClientSession
from ... base import ToolService
default_ident = "mcp-tool" default_ident = "mcp-tool"
class Service(ToolService): class Service(ToolService):
@ -19,14 +20,42 @@ class Service(ToolService):
**params **params
) )
self.register_config_handler(self.on_mcp_config)
self.mcp_services = {}
async def on_mcp_config(self, config, version):
print("Got config version", version)
if "mcp" not in config: return
self.mcp_services = {
k: json.loads(v)
for k, v in config["mcp"].items()
}
async def invoke_tool(self, name, parameters): async def invoke_tool(self, name, parameters):
try: try:
print(name)
print(parameters) if name not in self.mcp_services:
raise RuntimeError(f"MCP service {name} not known")
if "url" not in self.mcp_services[name]:
raise RuntimeError(f"MCP service {name} URL not defined")
url = self.mcp_services[name]["url"]
if "name" in self.mcp_services[name]:
remote_name = self.mcp_services[name]["name"]
else:
remote_name = name
print("Invoking", remote_name, "at", url, flush=True)
# Connect to a streamable HTTP server # Connect to a streamable HTTP server
# async with streamablehttp_client("http://mcp-server:8000/mcp/") as ( async with streamablehttp_client(url) as (
async with streamablehttp_client("http://host.containers.internal:9870/mcp/") as (
read_stream, read_stream,
write_stream, write_stream,
_, _,
@ -40,12 +69,10 @@ class Service(ToolService):
# Call a tool # Call a tool
result = await session.call_tool( result = await session.call_tool(
name, remote_name,
parameters parameters
) )
print(result)
if result.structuredContent: if result.structuredContent:
return result.structuredContent return result.structuredContent
elif hasattr(result, "content"): elif hasattr(result, "content"):