External Agent Integration
Integrating External Agents into Crowe
Crowe allows seamless integration of agents from other frameworks — including LangChain, Griptape, Hugging Face, Rasa, DialogFlow, and more — so you can orchestrate diverse AI systems under a single, unified agent workflow.
This section provides step-by-step guides to bring external agents into Crowe by creating new agent classes, implementing the required methods, and ensuring full compatibility.
Quick Overview
Create a custom class inheriting from
Agentin Crowe.Override the
.run(task: str) -> strmethod to execute the external agent.(Optional) Add helper methods for saving output (JSON, DB, logs, etc.).
Crowe Agent Class Template
The foundation for any external integration is the Agent base class from Crowe.
from crowe import Agent
import json
from pathlib import Path
class ExternalCroweAgent(Agent):
def run(self, task: str) -> str:
# TODO: Replace with actual external agent logic
return f"[ExternalCroweAgent] Received task: {task}"
def save_to_json(self, output: str, filepath: str):
try:
Path(filepath).parent.mkdir(parents=True, exist_ok=True)
with open(filepath, "w", encoding="utf-8") as f:
json.dump({"response": output}, f, ensure_ascii=False, indent=2)
except Exception as e:
print(f"[ExternalCroweAgent] Failed to save JSON: {e}")
# Usage
agent = ExternalCroweAgent()
result = agent.run("Analyze dataset trends")
agent.save_to_json(result, "outputs/result.json")
print(result)Example 1 – Griptape Integration
Griptape provides tools for web scraping, summarization, and file management. Here’s how to wrap a Griptape agent into Crowe:
from __future__ import annotations
import json
import logging
import re
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
from crowe import Agent as CroweAgent
from griptape.structures import Agent as GriptapeAgent
from griptape.tools import WebScraperTool, FileManagerTool, PromptSummaryTool
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)
class GriptapeCroweAgent(CroweAgent):
"""
Crowe-compatible wrapper around a Griptape Agent.
Features
- Accepts task as "url, outfile" string or {"url": "...", "outfile": "..."} dict
- Validates URL and normalizes output filename
- Saves artifact via Griptape FileManagerTool or local filesystem fallback
- Returns text or JSON payload
- Optional batch mode (run_many)
"""
DEFAULT_TEMPLATE = (
"Load {{ args[0] }}, summarize the key ideas in concise bullet points, "
"and store the result in {{ args[1] }}."
)
def __init__(
self,
*,
prompt_template: Optional[str] = None,
default_out_ext: str = ".txt",
save_dir: Union[str, Path] = "artifacts",
return_format: str = "text", # "text" | "json"
griptape_tools: Optional[List[Any]] = None,
**kwargs: Any,
):
"""
:param prompt_template: Jinja template used by Griptape agent.
:param default_out_ext: Ensures outfile has an extension.
:param save_dir: Local fallback directory for saving outputs.
:param return_format: Return "text" (string) or "json" ({url, outfile, content}).
:param griptape_tools: Custom tool list; falls back to (WebScraper, PromptSummary, FileManager).
:param kwargs: Crowe Agent kwargs.
"""
super().__init__(**kwargs)
self.default_out_ext = default_out_ext if default_out_ext.startswith(".") else f".{default_out_ext}"
self.save_dir = Path(save_dir)
self.save_dir.mkdir(parents=True, exist_ok=True)
self.return_format = return_format
tools = griptape_tools or [
WebScraperTool(off_prompt=True),
PromptSummaryTool(off_prompt=True),
FileManagerTool(),
]
# Initialize the underlying Griptape agent
self._gt = GriptapeAgent(
input=(prompt_template or self.DEFAULT_TEMPLATE),
tools=tools,
)
# -----------------------
# Public Crowe API
# -----------------------
def run(self, task: Union[str, Dict[str, Any]]) -> str:
"""
Execute a single scrape→summarize flow.
:param task: "url, outfile" or {"url": "...", "outfile": "..."}
:return: str or JSON (as str) depending on self.return_format
"""
try:
url, outfile = self._parse_task(task)
logger.info(f"[GriptapeCroweAgent] Task parsed url={url}, outfile={outfile}")
# Delegate to Griptape
gt_result = self._gt.run(url, outfile)
# Try to read the artifact content; if FileManager stored it, fetch from disk as fallback
content = self._read_artifact(outfile) or str(gt_result)
payload = self._build_payload(url, outfile, content)
return payload if isinstance(payload, str) else json.dumps(payload, ensure_ascii=False, indent=2)
except Exception as e:
logger.exception("[GriptapeCroweAgent] Execution failed")
return json.dumps({"error": str(e)}, ensure_ascii=False)
def run_many(self, tasks: Iterable[Union[str, Dict[str, Any]]]) -> List[str]:
"""
Batch execution helper. Returns a list of serialized results (str/JSON-string).
"""
results: List[str] = []
for t in tasks:
results.append(self.run(t))
return results
# -----------------------
# Internals
# -----------------------
def _parse_task(self, task: Union[str, Dict[str, Any]]) -> Tuple[str, str]:
"""
Normalize task input into (url, outfile)
"""
if isinstance(task, str):
parts = [p.strip() for p in task.split(",") if p.strip()]
if len(parts) < 1:
raise ValueError("Task must contain at least a URL.")
url = parts[0]
outfile = parts[1] if len(parts) > 1 else "summary"
elif isinstance(task, dict):
url = str(task.get("url", "")).strip()
outfile = str(task.get("outfile", "summary")).strip()
else:
raise TypeError("Task must be a string or dict.")
self._validate_url(url)
outfile = self._normalize_outfile(outfile)
return url, outfile
@staticmethod
def _validate_url(url: str) -> None:
pattern = r"^https?://"
if not re.match(pattern, url):
raise ValueError(f"Invalid URL (must start with http/https): {url}")
def _normalize_outfile(self, name: str) -> str:
# strip path traversal and enforce extension
name = re.sub(r"[^\w\-.]+", "_", name).strip("_") or "summary"
if not name.lower().endswith(self.default_out_ext.lower()):
name = f"{name}{self.default_out_ext}"
return name
def _read_artifact(self, outfile: str) -> Optional[str]:
"""
Try loading the generated artifact if Griptape's FileManager wrote it locally.
"""
# common locations attempt
candidates = [
self.save_dir / outfile,
Path(outfile),
]
for c in candidates:
try:
if c.exists() and c.is_file():
return c.read_text(encoding="utf-8", errors="ignore")
except Exception:
continue
return None
def _build_payload(self, url: str, outfile: str, content: str) -> Union[str, Dict[str, Any]]:
if self.return_format == "json":
return {"url": url, "outfile": outfile, "content": content}
return content
# -----------------------
# Usage Examples
# -----------------------
if __name__ == "__main__":
# Example 1: simple string task
agent = GriptapeCroweAgent(return_format="json")
print(agent.run("https://example.com, example_summary"))
# Example 2: JSON task
job = {"url": "https://example.com/blog", "outfile": "blog_digest.md"}
print(agent.run(job))
# Example 3: batch
batch = [
"https://example.com/page-a, a.txt",
{"url": "https://example.com/page-b", "outfile": "b.txt"},
]
print(agent.run_many(batch))
Example 2 – LangChain Integration
LangChain excels at LLM orchestration. Here’s how to integrate it into Crowe:
from crowe import Agent as CroweAgent
from langchain import LLMChain
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
class LangChainCroweAgent(CroweAgent):
def __init__(self, temperature: float = 0.2, model: str = "gpt-4o-mini", *args, **kwargs):
super().__init__(*args, **kwargs)
prompt = PromptTemplate.from_template(
"You are concise and factual.\nQuestion: {q}\nAnswer:"
)
self.chain = LLMChain(
llm=OpenAI(model=model, temperature=temperature),
prompt=prompt,
)
def run(self, task: str) -> str:
try:
return self.chain.run({"q": task}).strip()
except Exception as e:
return f"[LangChainCroweAgent error] {e}"
# Usage
agent = LangChainCroweAgent()
print(agent.run("What is the capital of France?"))
Example 3 – OpenAI Function Calling
OpenAI models (like GPT-4) can call external functions for advanced tasks.
from crowe import Agent as CroweAgent
import openai
class OpenAIFunctionCroweAgent(CroweAgent):
def __init__(self, *args, **kwargs):
self.api_key = "YOUR_API_KEY"
super().__init__(*args, **kwargs)
def run(self, task: str) -> str:
command, text = task.split(", ")
response = openai.Completion.create(
model="gpt-4",
prompt=f"{command}: {text}",
temperature=0.5,
max_tokens=100,
)
return response.choices[0].text.strip()
# Usage
agent = OpenAIFunctionCroweAgent()
print(agent.run("summarize, This is a long paragraph..."))Example 4 – Hugging Face Transformers
Hugging Face offers pre-trained NLP models for text generation, QA, and more.
from crowe import Agent as CroweAgent
from transformers import pipeline
class HuggingFaceCroweAgent(CroweAgent):
def __init__(self, model_name: str, *args, **kwargs):
self.pipeline = pipeline("text-generation", model=model_name)
super().__init__(*args, **kwargs)
def run(self, task: str) -> str:
result = self.pipeline(task, max_length=50)
return result[0]["generated_text"]
# Usage
agent = HuggingFaceCroweAgent("gpt2")
print(agent.run("Once upon a time..."))Example 5 – Rasa Conversational Agent
Rasa is ideal for intent-based chatbots.
from crowe import Agent as CroweAgent
from rasa.core.agent import Agent as RasaAgent
class RasaCroweAgent(CroweAgent):
def __init__(self, model_path: str, *args, **kwargs):
self.agent = RasaAgent.load(model_path)
super().__init__(*args, **kwargs)
def run(self, task: str) -> str:
result = self.agent.handle_text(task)
return result[0]["text"] if result else "No response."
# Usage
agent = RasaCroweAgent("path/to/rasa_model")
print(agent.run("Hello, how can I get a refund?"))Example 6 – DialogFlow Integration
Google’s DialogFlow is useful for multi-turn conversations.
from crowe import Agent as CroweAgent
from google.cloud import dialogflow_v2 as dialogflow # v2 API
from typing import Optional
class DialogFlowCroweAgent(CroweAgent):
def __init__(
self,
project_id: str,
session_id: str,
language_code: str = "en-US",
*args, **kwargs
):
super().__init__(*args, **kwargs)
self.client = dialogflow.SessionsClient()
self.session = self.client.session_path(project_id, session_id)
self.language_code = language_code
def run(self, task: str) -> str:
try:
text_input = dialogflow.TextInput(text=task, language_code=self.language_code)
query_input = dialogflow.QueryInput(text=text_input)
resp = self.client.detect_intent(request={"session": self.session, "query_input": query_input})
return (resp.query_result.fulfillment_text or "").strip() or "[No fulfillment text]"
except Exception as e:
return f"[DialogFlowCroweAgent error] {e}"
# Optional helper: send with a context hint
def run_with_context(self, task: str, context_name: str, lifespan: int = 2) -> str:
try:
text_input = dialogflow.TextInput(text=task, language_code=self.language_code)
query_input = dialogflow.QueryInput(text=text_input)
context = dialogflow.Context(
name=f"{self.client.project_agent_session_path(*self.session.split('/')[-4:-2])}/contexts/{context_name}",
lifespan_count=lifespan,
)
resp = self.client.detect_intent(
request={
"session": self.session,
"query_input": query_input,
"query_params": {"contexts": [context]},
}
)
return (resp.query_result.fulfillment_text or "").strip() or "[No fulfillment text]"
except Exception as e:
return f"[DialogFlowCroweAgent error] {e}"
# Usage
agent = DialogFlowCroweAgent("my_project", "session_123")
print(agent.run("Book me a flight to Paris."))Example 7 – Custom API Agent
Crowe can wrap any REST/GraphQL API as an agent.
from crowe import Agent as CroweAgent
import requests
class APICroweAgent(CroweAgent):
def run(self, task: str) -> str:
try:
endpoint, query = [x.strip() for x in task.split(",", 1)]
resp = requests.get(endpoint, params={"q": query}, timeout=10)
resp.raise_for_status()
return resp.text.strip()
except Exception as e:
return f"[APICroweAgent error] {e}"
# Usage
agent = APICroweAgent()
print(agent.run("https://api.example.com/search, python"))Summary of Crowe Integrations
Griptape → Web scraping + summarization workflows.
LangChain → LLM orchestration.
OpenAI Function Calling → Execute external APIs and functions.
Rasa → Intent-driven chatbots.
Hugging Face → Pre-trained transformer models.
DialogFlow → Multi-turn conversation flows.
Custom APIs → Any REST/GraphQL service as an agent.
Conclusion
Crowe’s modular Agent API makes it easy to plug in AI systems from any ecosystem. By following the above patterns, you can bring together multiple AI frameworks into a single, orchestrated environment, unlocking hybrid workflows that combine the strengths of each platform.
Last updated