mirror of
https://github.com/trustgraph-ai/trustgraph.git
synced 2026-07-02 02:58:10 +02:00
MCP client support (#427)
- MCP client service - Tool request/response schema - API gateway support for mcp-tool - Message translation for tool request & response - Make mcp-tool using configuration service for information about where the MCP services are.
This commit is contained in:
parent
21bee4cd83
commit
e56186054a
13 changed files with 356 additions and 2 deletions
51
trustgraph-base/trustgraph/messaging/translators/tool.py
Normal file
51
trustgraph-base/trustgraph/messaging/translators/tool.py
Normal file
|
|
@ -0,0 +1,51 @@
|
|||
import json
|
||||
from typing import Dict, Any, Tuple
|
||||
from ...schema import ToolRequest, ToolResponse
|
||||
from .base import MessageTranslator
|
||||
|
||||
class ToolRequestTranslator(MessageTranslator):
|
||||
"""Translator for ToolRequest schema objects"""
|
||||
|
||||
def to_pulsar(self, data: Dict[str, Any]) -> ToolRequest:
|
||||
# Handle both "name" and "parameters" input keys
|
||||
name = data.get("name", "")
|
||||
if "parameters" in data:
|
||||
parameters = json.dumps(data["parameters"])
|
||||
else:
|
||||
parameters = None
|
||||
|
||||
return ToolRequest(
|
||||
name = name,
|
||||
parameters = parameters,
|
||||
)
|
||||
|
||||
def from_pulsar(self, obj: ToolRequest) -> Dict[str, Any]:
|
||||
result = {}
|
||||
|
||||
if obj.name:
|
||||
result["name"] = obj.name
|
||||
if obj.parameters is not None:
|
||||
result["parameters"] = json.loads(obj.parameters)
|
||||
|
||||
return result
|
||||
|
||||
class ToolResponseTranslator(MessageTranslator):
|
||||
"""Translator for ToolResponse schema objects"""
|
||||
|
||||
def to_pulsar(self, data: Dict[str, Any]) -> ToolResponse:
|
||||
raise NotImplementedError("Response translation to Pulsar not typically needed")
|
||||
|
||||
def from_pulsar(self, obj: ToolResponse) -> Dict[str, Any]:
|
||||
|
||||
result = {}
|
||||
|
||||
if obj.text:
|
||||
result["text"] = obj.text
|
||||
if obj.object:
|
||||
result["object"] = json.loads(obj.object)
|
||||
|
||||
return result
|
||||
|
||||
def from_response_with_completion(self, obj: ToolResponse) -> Tuple[Dict[str, Any], bool]:
|
||||
"""Returns (response_dict, is_final)"""
|
||||
return self.from_pulsar(obj), True
|
||||
Loading…
Add table
Add a link
Reference in a new issue