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

Specify request/response queues

Browse source
This commit is contained in:
Cyber MacGeddon 2026-03-04 14:25:06 +00:00
parent ade19c7d4b
commit 5f53cf1848
3 changed files with 46 additions and 49 deletions
Download patch file Download diff file Expand all files Collapse all files

12
trustgraph-flow/trustgraph/agent/react/service.py
View file

@ -203,10 +203,11 @@ class Processor(AgentService):
)
service_config = tool_services[service_ref]
service_topic = service_config.get("topic")
if not service_topic:
request_queue = service_config.get("request-queue")
response_queue = service_config.get("response-queue")
if not request_queue or not response_queue:
raise RuntimeError(
f"Tool-service '{service_ref}' has no 'topic' defined"
f"Tool-service '{service_ref}' must define 'request-queue' and 'response-queue'"
)
# Build config values from tool config
@ -233,10 +234,11 @@ class Processor(AgentService):
for arg in config_args
]
# Store topic for the implementation
# Store queues for the implementation
impl = functools.partial(
ToolServiceImpl,
service_topic=service_topic,
request_queue=request_queue,
response_queue=response_queue,
config_values=config_values,
arguments=arguments,
processor=self,

45
trustgraph-flow/trustgraph/agent/react/tools.py
View file

@ -210,23 +210,25 @@ class ToolServiceImpl:
Implementation for dynamically pluggable tool services.
Tool services are external Pulsar services that can be invoked as agent tools.
The service is configured via a tool-service descriptor that defines the topic,
The service is configured via a tool-service descriptor that defines the queues,
and a tool descriptor that provides config values and argument definitions.
"""
def __init__(self, context, service_topic, config_values=None, arguments=None, processor=None):
def __init__(self, context, request_queue, response_queue, config_values=None, arguments=None, processor=None):
"""
Initialize a tool service implementation.
Args:
context: The context function (provides user info)
service_topic: The base topic name for the service
request_queue: Full Pulsar topic for requests
response_queue: Full Pulsar topic for responses
config_values: Dict of config values (e.g., {"collection": "customers"})
arguments: List of Argument objects defining the tool's parameters
processor: The Processor instance (for pubsub access)
"""
self.context = context
self.service_topic = service_topic
self.request_queue = request_queue
self.response_queue = response_queue
self.config_values = config_values or {}
self.arguments = arguments or []
self.processor = processor
@ -240,9 +242,10 @@ class ToolServiceImpl:
if self._client is not None:
return self._client
# Check if processor already has a client for this topic
if self.service_topic in self.processor.tool_service_clients:
self._client = self.processor.tool_service_clients[self.service_topic]
# Check if processor already has a client for this queue pair
client_key = f"{self.request_queue}|{self.response_queue}"
if client_key in self.processor.tool_service_clients:
self._client = self.processor.tool_service_clients[client_key]
return self._client
# Import here to avoid circular imports
@ -251,40 +254,28 @@ class ToolServiceImpl:
from trustgraph.schema import ToolServiceRequest, ToolServiceResponse
import uuid
# Construct full topic paths using non-persistent topics
# If topic already contains "://", assume it's a full Pulsar URI
# Otherwise, construct default path
if '://' in self.service_topic:
# Full Pulsar URI provided - use as base for request/response
request_topic = self.service_topic.replace("://tg/", "://tg/request/")
response_topic = self.service_topic.replace("://tg/", "://tg/response/")
else:
# Construct default path matching DynamicToolService pattern
request_topic = f"non-persistent://tg/request/{self.service_topic}"
response_topic = f"non-persistent://tg/response/{self.service_topic}"
request_metrics = ProducerMetrics(
processor=self.processor.id,
flow="tool-service",
name=request_topic
name=self.request_queue
)
response_metrics = SubscriberMetrics(
processor=self.processor.id,
flow="tool-service",
name=response_topic
name=self.response_queue
)
# Create unique subscription for responses
subscription = f"{self.processor.id}--tool-service--{self.service_topic}--{uuid.uuid4()}"
subscription = f"{self.processor.id}--tool-service--{uuid.uuid4()}"
self._client = ToolServiceClient(
backend=self.processor.pubsub,
subscription=subscription,
consumer_name=self.processor.id,
request_topic=request_topic,
request_topic=self.request_queue,
request_schema=ToolServiceRequest,
request_metrics=request_metrics,
response_topic=response_topic,
response_topic=self.response_queue,
response_schema=ToolServiceResponse,
response_metrics=response_metrics,
)
@ -293,13 +284,13 @@ class ToolServiceImpl:
await self._client.start()
# Register for cleanup
self.processor.tool_service_clients[self.service_topic] = self._client
self.processor.tool_service_clients[client_key] = self._client
logger.debug(f"Created tool service client for {self.service_topic}")
logger.debug(f"Created tool service client for {self.request_queue}")
return self._client
async def invoke(self, **arguments):
logger.debug(f"Tool service invocation: {self.service_topic}...")
logger.debug(f"Tool service invocation: {self.request_queue}...")
logger.debug(f"Config: {self.config_values}")
logger.debug(f"Arguments: {arguments}")