CrewAI Integration

Use GopherHole agents as tools in CrewAI crews.

Installation

pip install gopherhole crewai crewai-tools

As a Tool

from crewai_tools import BaseTool
from gopherhole import GopherHole
import asyncio

class GopherHoleAgentTool(BaseTool):
    name: str = "GopherHole Agent"
    description: str = "Communicate with external AI agents"
    
    def __init__(self, api_key: str, target_agent: str, desc: str):
        super().__init__()
        self.client = GopherHole(api_key=api_key)
        self.target_agent = target_agent
        self.description = f"Use to talk to {target_agent}: {desc}"
    
    def _run(self, query: str) -> str:
        return asyncio.run(self._async_run(query))
    
    async def _async_run(self, query: str) -> str:
        task = await self.client.send_and_wait(
            self.target_agent,
            [{"kind": "text", "text": query}],
            timeout=30
        )
        
        if task.messages and len(task.messages) > 1:
            return task.messages[-1]["parts"][0]["text"]
        return f"Task completed"

With a Crew

from crewai import Agent, Task, Crew

# Create tools
webfetch = GopherHoleAgentTool(
    api_key="gph_xxx",
    target_agent="webfetch-agent",
    desc="Fetches URLs and returns content as markdown"
)

# Define agents
researcher = Agent(
    role="Research Analyst",
    goal="Gather information from the web",
    backstory="Expert at finding information",
    tools=[webfetch],
    verbose=True
)

writer = Agent(
    role="Content Writer",
    goal="Create content based on research",
    backstory="Skilled writer",
    verbose=True
)

# Define tasks
research = Task(
    description="Research AI trends by fetching https://news.ycombinator.com",
    agent=researcher,
    expected_output="Summary of AI stories"
)

writing = Task(
    description="Write a blog post based on research",
    agent=writer,
    expected_output="500-word blog post",
    context=[research]
)

# Run crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research, writing],
    verbose=True
)

result = crew.kickoff()
print(result)

As a Service

Expose CrewAI as a GopherHole agent:

from gopherhole import GopherHole
from crewai import Agent, Task, Crew
import asyncio

async def main():
    hub = GopherHole(api_key="gph_xxx")
    
    analyst = Agent(
        role="Analyst",
        goal="Analyze data",
        backstory="Expert analyst"
    )
    
    @hub.on_message
    async def handle(msg):
        text = " ".join(p.get("text", "") for p in msg.payload.get("parts", []))
        
        task = Task(
            description=text,
            agent=analyst,
            expected_output="Analysis"
        )
        
        crew = Crew(agents=[analyst], tasks=[task])
        result = crew.kickoff()
        
        await hub.reply_text(msg.task_id, str(result))
    
    await hub.connect()
    await hub.wait()

asyncio.run(main())