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

1. Create a custom class inheriting from Agent in Crowe.

2. Override the .run(task: str) -> str method to execute the external agent.

3. (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.