From 5f53cf18481aaeed851978852f21e58f3af80d63 Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Wed, 4 Mar 2026 14:25:06 +0000 Subject: [PATCH] Specify request/response queues --- docs/tech-specs/tool-services.md | 38 +++++++++------- .../trustgraph/agent/react/service.py | 12 ++--- .../trustgraph/agent/react/tools.py | 45 ++++++++----------- 3 files changed, 46 insertions(+), 49 deletions(-) diff --git a/docs/tech-specs/tool-services.md b/docs/tech-specs/tool-services.md index 17c7bf7e..20e004d3 100644 --- a/docs/tech-specs/tool-services.md +++ b/docs/tech-specs/tool-services.md @@ -62,13 +62,14 @@ elif impl_id == "text-completion": #### Tier 1: Tool Service Descriptor A tool service defines a Pulsar service interface. It declares: -- The topic to call +- The Pulsar queues for request/response - Configuration parameters it requires from tools that use it ```json { "id": "custom-rag", - "topic": "custom-rag-request", + "request-queue": "non-persistent://tg/request/custom-rag", + "response-queue": "non-persistent://tg/response/custom-rag", "config-params": [ {"name": "collection", "required": true} ] @@ -80,7 +81,8 @@ A tool service that needs no configuration parameters: ```json { "id": "calculator", - "topic": "calc-request", + "request-queue": "non-persistent://tg/request/calc", + "response-queue": "non-persistent://tg/response/calc", "config-params": [] } ``` @@ -199,16 +201,18 @@ Requests and responses use untyped dicts. No schema validation at the agent leve ### Client Interface: Direct Pulsar Topics -Tool services use direct Pulsar topics without requiring flow configuration: +Tool services use direct Pulsar topics without requiring flow configuration. The tool-service descriptor specifies the full queue names: -- **Request Topic**: `non-persistent://tg/request/{topic}` -- **Response Topic**: `non-persistent://tg/response/{topic}` +```json +{ + "id": "joke-service", + "request-queue": "non-persistent://tg/request/joke", + "response-queue": "non-persistent://tg/response/joke", + "config-params": [...] +} +``` -Where `{topic}` comes from the tool-service descriptor. For example, a service with `"topic": "joke"` uses: -- `non-persistent://tg/request/joke` -- `non-persistent://tg/response/joke` - -This avoids the complexity of flow blueprints and allows services to be deployed independently. +This allows services to be hosted in any namespace. ### Error Handling: Standard Error Convention @@ -295,8 +299,7 @@ class DynamicToolService(AsyncProcessor): def __init__(self, **params): topic = params.get("topic", default_topic) - request_topic = f"non-persistent://tg/request/{topic}" - response_topic = f"non-persistent://tg/response/{topic}" + # Constructs topics: non-persistent://tg/request/{topic}, non-persistent://tg/response/{topic} # Sets up Consumer and Producer async def invoke(self, user, config, arguments): @@ -310,8 +313,8 @@ Implementation in `trustgraph-flow/trustgraph/agent/react/tools.py`: ```python class ToolServiceImpl: - def __init__(self, context, service_topic, config_values, arguments, processor): - # Constructs topic paths: non-persistent://tg/request/{topic}, non-persistent://tg/response/{topic} + def __init__(self, context, request_queue, response_queue, config_values, arguments, processor): + # Uses the provided queue paths directly # Creates ToolServiceClient on first use async def invoke(self, **arguments): @@ -326,7 +329,7 @@ Two config sections: ``` tool-service/ - joke-service: {"id": "joke-service", "topic": "joke", "config-params": [...]} + joke-service: {"id": "joke-service", "request-queue": "non-persistent://tg/request/joke", "response-queue": "non-persistent://tg/response/joke", "config-params": [...]} tool/ tell-joke: {"type": "tool-service", "service": "joke-service", ...} @@ -359,7 +362,8 @@ Tool service config: ```json { "id": "joke-service", - "topic": "joke", + "request-queue": "non-persistent://tg/request/joke", + "response-queue": "non-persistent://tg/response/joke", "config-params": [{"name": "style", "required": false}] } ``` diff --git a/trustgraph-flow/trustgraph/agent/react/service.py b/trustgraph-flow/trustgraph/agent/react/service.py index c4cb1fab..1c96adef 100755 --- a/trustgraph-flow/trustgraph/agent/react/service.py +++ b/trustgraph-flow/trustgraph/agent/react/service.py @@ -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, diff --git a/trustgraph-flow/trustgraph/agent/react/tools.py b/trustgraph-flow/trustgraph/agent/react/tools.py index 902ad685..18675084 100644 --- a/trustgraph-flow/trustgraph/agent/react/tools.py +++ b/trustgraph-flow/trustgraph/agent/react/tools.py @@ -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}")