diff --git a/trustgraph-embeddings-hf/trustgraph/embeddings/hf/hf.py b/trustgraph-embeddings-hf/trustgraph/embeddings/hf/hf.py
index 5470333b..2e44821e 100755
--- a/trustgraph-embeddings-hf/trustgraph/embeddings/hf/hf.py
+++ b/trustgraph-embeddings-hf/trustgraph/embeddings/hf/hf.py
@@ -56,7 +56,7 @@ class Processor(ConsumerProducer):
 
             print("Send response...", flush=True)
             r = EmbeddingsResponse(vectors=embeds, error=None)
-            self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
             print("Done.", flush=True)
 
@@ -75,7 +75,7 @@ class Processor(ConsumerProducer):
                 response=None,
             )
 
-            self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
             self.consumer.acknowledge(msg)
             
diff --git a/trustgraph-flow/trustgraph/agent/react/service.py b/trustgraph-flow/trustgraph/agent/react/service.py
index 2df0fa4d..bc045b71 100755
--- a/trustgraph-flow/trustgraph/agent/react/service.py
+++ b/trustgraph-flow/trustgraph/agent/react/service.py
@@ -231,7 +231,7 @@ class Processor(ConsumerProducer):
                     observation=None,
                 )
 
-                await self.producer.send(r, properties={"id": id})
+                await self.send(r, properties={"id": id})
 
             async def observe(x):
 
@@ -244,7 +244,7 @@ class Processor(ConsumerProducer):
                     observation=x,
                 )
 
-                await self.producer.send(r, properties={"id": id})
+                await self.send(r, properties={"id": id})
 
             act = self.agent.react(v.question, history, think, observe)
 
@@ -260,7 +260,7 @@ class Processor(ConsumerProducer):
                     thought=None,
                 )
 
-                await self.producer.send(r, properties={"id": id})
+                await self.send(r, properties={"id": id})
 
                 print("Done.", flush=True)
 
@@ -283,7 +283,7 @@ class Processor(ConsumerProducer):
                 ]
             )
 
-            await self.recursive_input.send(r, properties={"id": id})
+            self.recursive_input.send(r, properties={"id": id})
 
             print("Done.", flush=True)
 
@@ -303,7 +303,7 @@ class Processor(ConsumerProducer):
                 response=None,
             )
 
-            await self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
     @staticmethod
     def add_args(parser):
diff --git a/trustgraph-flow/trustgraph/embeddings/fastembed/processor.py b/trustgraph-flow/trustgraph/embeddings/fastembed/processor.py
index 253d77ac..bc164fa0 100755
--- a/trustgraph-flow/trustgraph/embeddings/fastembed/processor.py
+++ b/trustgraph-flow/trustgraph/embeddings/fastembed/processor.py
@@ -65,7 +65,7 @@ class Processor(ConsumerProducer):
             error=None,
         )
 
-        await self.producer.send(r, properties={"id": id})
+        await self.send(r, properties={"id": id})
 
         print("Done.", flush=True)
 
diff --git a/trustgraph-flow/trustgraph/embeddings/ollama/processor.py b/trustgraph-flow/trustgraph/embeddings/ollama/processor.py
index f9dd87a9..c441b9c6 100755
--- a/trustgraph-flow/trustgraph/embeddings/ollama/processor.py
+++ b/trustgraph-flow/trustgraph/embeddings/ollama/processor.py
@@ -67,7 +67,7 @@ class Processor(ConsumerProducer):
             error=None,
         )
 
-        await self.producer.send(r, properties={"id": id})
+        await self.send(r, properties={"id": id})
 
         print("Done.", flush=True)
 
diff --git a/trustgraph-flow/trustgraph/external/wikipedia/service.py b/trustgraph-flow/trustgraph/external/wikipedia/service.py
index a269c8ce..cc002765 100644
--- a/trustgraph-flow/trustgraph/external/wikipedia/service.py
+++ b/trustgraph-flow/trustgraph/external/wikipedia/service.py
@@ -60,7 +60,7 @@ class Processor(ConsumerProducer):
                 text=resp
             )
 
-            await self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
             self.consumer.acknowledge(msg)
 
@@ -75,7 +75,7 @@ class Processor(ConsumerProducer):
                 ),
                 text=None,
             )
-            await self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
             self.consumer.acknowledge(msg)
 
diff --git a/trustgraph-flow/trustgraph/extract/object/row/extract.py b/trustgraph-flow/trustgraph/extract/object/row/extract.py
index db02a651..9ccf3370 100755
--- a/trustgraph-flow/trustgraph/extract/object/row/extract.py
+++ b/trustgraph-flow/trustgraph/extract/object/row/extract.py
@@ -130,14 +130,14 @@ class Processor(ConsumerProducer):
         t = Rows(
             metadata=metadata, row_schema=self.row_schema, rows=rows
         )
-        await self.producer.send(t)
+        await self.send(t)
 
     def emit_vec(self, metadata, name, vec, key_name, key):
 
         r = ObjectEmbeddings(
             metadata=metadata, vectors=vec, name=name, key_name=key_name, id=key
         )
-        await self.vec_prod.send(r)
+        self.vec_prod.send(r)
 
     async def handle(self, msg):
 
diff --git a/trustgraph-flow/trustgraph/model/prompt/generic/service.py b/trustgraph-flow/trustgraph/model/prompt/generic/service.py
index 8d533fcf..b143b759 100755
--- a/trustgraph-flow/trustgraph/model/prompt/generic/service.py
+++ b/trustgraph-flow/trustgraph/model/prompt/generic/service.py
@@ -164,7 +164,7 @@ class Processor(ConsumerProducer):
 
             print("Send response...", flush=True)
             r = PromptResponse(definitions=output, error=None)
-            await self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
             print("Done.", flush=True)
         
@@ -182,7 +182,7 @@ class Processor(ConsumerProducer):
                 response=None,
             )
 
-            await self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
     async def handle_extract_topics(self, id, v):
 
@@ -223,7 +223,7 @@ class Processor(ConsumerProducer):
 
             print("Send response...", flush=True)
             r = PromptResponse(topics=output, error=None)
-            await self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
             print("Done.", flush=True)
         
@@ -241,7 +241,7 @@ class Processor(ConsumerProducer):
                 response=None,
             )
 
-            await self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
     async def handle_extract_relationships(self, id, v):
 
@@ -295,7 +295,7 @@ class Processor(ConsumerProducer):
 
             print("Send response...", flush=True)
             r = PromptResponse(relationships=output, error=None)
-            await self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
             print("Done.", flush=True)
 
@@ -313,7 +313,7 @@ class Processor(ConsumerProducer):
                 response=None,
             )
 
-            await self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
     async def handle_extract_rows(self, id, v):
 
@@ -366,7 +366,7 @@ class Processor(ConsumerProducer):
 
             print("Send response...", flush=True)
             r = PromptResponse(rows=output, error=None)
-            await self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
             print("Done.", flush=True)
 
@@ -384,7 +384,7 @@ class Processor(ConsumerProducer):
                 response=None,
             )
 
-            await self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
         
     async def handle_kg_prompt(self, id, v):
 
@@ -400,7 +400,7 @@ class Processor(ConsumerProducer):
 
             print("Send response...", flush=True)
             r = PromptResponse(answer=ans, error=None)
-            await self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
             print("Done.", flush=True)
 
@@ -418,7 +418,7 @@ class Processor(ConsumerProducer):
                 response=None,
             )
 
-            await self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
         
     async def handle_document_prompt(self, id, v):
 
@@ -437,7 +437,7 @@ class Processor(ConsumerProducer):
 
             print("Send response...", flush=True)
             r = PromptResponse(answer=ans, error=None)
-            await self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
             print("Done.", flush=True)
 
@@ -455,7 +455,7 @@ class Processor(ConsumerProducer):
                 response=None,
             )
 
-            await self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
         
     @staticmethod
     def add_args(parser):
diff --git a/trustgraph-flow/trustgraph/model/text_completion/azure/llm.py b/trustgraph-flow/trustgraph/model/text_completion/azure/llm.py
index 8951b71e..33840378 100755
--- a/trustgraph-flow/trustgraph/model/text_completion/azure/llm.py
+++ b/trustgraph-flow/trustgraph/model/text_completion/azure/llm.py
@@ -154,7 +154,7 @@ class Processor(ConsumerProducer):
             print("Send response...", flush=True)
 
             r = TextCompletionResponse(response=resp, error=None, in_token=inputtokens, out_token=outputtokens, model=self.model)
-            await self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
         except TooManyRequests:
 
@@ -182,7 +182,7 @@ class Processor(ConsumerProducer):
                 model=None,
             )
 
-            await self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
             self.consumer.acknowledge(msg)
 
diff --git a/trustgraph-flow/trustgraph/model/text_completion/azure_openai/llm.py b/trustgraph-flow/trustgraph/model/text_completion/azure_openai/llm.py
index bce44ab6..252d58ad 100755
--- a/trustgraph-flow/trustgraph/model/text_completion/azure_openai/llm.py
+++ b/trustgraph-flow/trustgraph/model/text_completion/azure_openai/llm.py
@@ -135,7 +135,7 @@ class Processor(ConsumerProducer):
                 model=self.model
             )
 
-            await self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
         except RateLimitError:
 
@@ -163,7 +163,7 @@ class Processor(ConsumerProducer):
                 model=None,
             )
 
-            await self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
             self.consumer.acknowledge(msg)
 
diff --git a/trustgraph-flow/trustgraph/model/text_completion/cohere/llm.py b/trustgraph-flow/trustgraph/model/text_completion/cohere/llm.py
index 4b18f7f1..d5dab142 100755
--- a/trustgraph-flow/trustgraph/model/text_completion/cohere/llm.py
+++ b/trustgraph-flow/trustgraph/model/text_completion/cohere/llm.py
@@ -136,7 +136,7 @@ class Processor(ConsumerProducer):
                 model=None,
             )
 
-            await self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
             self.consumer.acknowledge(msg)
 
diff --git a/trustgraph-vertexai/trustgraph/model/text_completion/vertexai/llm.py b/trustgraph-vertexai/trustgraph/model/text_completion/vertexai/llm.py
index efe63ea6..4d38c8c0 100755
--- a/trustgraph-vertexai/trustgraph/model/text_completion/vertexai/llm.py
+++ b/trustgraph-vertexai/trustgraph/model/text_completion/vertexai/llm.py
@@ -169,7 +169,7 @@ class Processor(ConsumerProducer):
                 model=self.model
             )
 
-            self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
             print("Done.", flush=True)
 
@@ -202,7 +202,7 @@ class Processor(ConsumerProducer):
                 model=None,
             )
 
-            self.producer.send(r, properties={"id": id})
+            await self.send(r, properties={"id": id})
 
             self.consumer.acknowledge(msg)