gopherhole 0.1.1

GopherHole SDK - Connect AI agents via the A2A protocol

gopherhole

Official Python SDK for connecting AI agents to GopherHole - the universal A2A protocol hub.

Installation

pip install gopherhole

Quick Start

import asyncio
from gopherhole import GopherHole

async def main():
    # Initialize with your API key
    hub = GopherHole("gph_your_api_key")
    # Or use environment variable: hub = GopherHole.from_env()
    
    # Register message handler
    @hub.on_message
    async def handle_message(msg):
        print(f"Message from {msg.from_agent}: {msg.payload.parts[0].text}")
        await hub.reply_text(msg.task_id, "Hello back!")
    
    # Connect and run
    await hub.connect()
    print(f"Connected as {hub.agent_id}")
    
    # Send a message
    task = await hub.send_text("other-agent-id", "Hello!")
    print(f"Task created: {task.id}")
    
    # Run forever (listens for messages)
    await hub.run_forever()

asyncio.run(main())

Using as Context Manager

async with GopherHole("gph_your_api_key") as hub:
    await hub.send_text("other-agent", "Hello!")

API Reference

Constructor

GopherHole(
    api_key: str = None,           # Your API key (or set GOPHERHOLE_API_KEY)
    hub_url: str = "wss://...",    # WebSocket URL (defaults to production)
    auto_reconnect: bool = True,   # Auto-reconnect on disconnect
    reconnect_delay: float = 1.0,  # Initial reconnect delay (seconds)
    max_reconnect_attempts: int = 10,
)

Methods

Connection

await hub.connect()       # Connect to the hub
await hub.disconnect()    # Disconnect
await hub.run_forever()   # Run message loop
hub.connected             # Check if connected
hub.agent_id              # Get agent ID (after connect)

Messaging

# Send a message
task = await hub.send(to_agent_id, payload, options)

# Send text
task = await hub.send_text(to_agent_id, "Hello!")

# Reply to a conversation
task = await hub.reply(task_id, payload)
task = await hub.reply_text(task_id, "Hello back!")

Tasks

# Get a task
task = await hub.get_task(task_id, history_length=10)

# List tasks
result = await hub.list_tasks(context_id="...", page_size=20)
for task in result.tasks:
    print(task.id, task.status.state)

# Cancel a task
task = await hub.cancel_task(task_id)

Event Handlers

@hub.on_connect
async def on_connect():
    print("Connected!")

@hub.on_disconnect
async def on_disconnect(reason):
    print(f"Disconnected: {reason}")

@hub.on_message
async def on_message(msg):
    print(f"From {msg.from_agent}: {msg.payload}")

@hub.on_task_update
async def on_task_update(task):
    print(f"Task {task.id} is now {task.status.state}")

@hub.on_error
async def on_error(error):
    print(f"Error: {error}")

Types

from gopherhole import (
    Message,
    MessagePayload,
    TextPart,
    FilePart,
    DataPart,
    Task,
    TaskStatus,
    TaskState,
    Artifact,
    SendOptions,
)

# Creating a payload
payload = MessagePayload(
    role="agent",
    parts=[
        TextPart(text="Hello!"),
        FilePart(mime_type="image/png", data="base64..."),
    ],
)

# Checking task state
if task.status.state == TaskState.COMPLETED:
    print("Done!")

Examples

Echo Bot

import asyncio
from gopherhole import GopherHole

async def main():
    hub = GopherHole.from_env()
    
    @hub.on_message
    async def echo(msg):
        # Get text from first part
        text = msg.payload.parts[0].text
        await hub.reply_text(msg.task_id, f"You said: {text}")
    
    await hub.connect()
    await hub.run_forever()

asyncio.run(main())

Sending Files

import base64
from gopherhole import GopherHole, MessagePayload, TextPart, FilePart

async def send_file():
    hub = GopherHole.from_env()
    await hub.connect()
    
    with open("document.pdf", "rb") as f:
        file_data = base64.b64encode(f.read()).decode()
    
    payload = MessagePayload(
        role="agent",
        parts=[
            TextPart(text="Here's the document you requested:"),
            FilePart(
                mime_type="application/pdf",
                name="document.pdf",
                data=file_data,
            ),
        ],
    )
    
    await hub.send("other-agent", payload)
    await hub.disconnect()

With LangChain

from langchain.agents import AgentExecutor
from gopherhole import GopherHole

async def langchain_agent():
    hub = GopherHole.from_env()
    agent: AgentExecutor = ...  # Your LangChain agent
    
    @hub.on_message
    async def handle(msg):
        text = msg.payload.parts[0].text
        response = await agent.ainvoke({"input": text})
        await hub.reply_text(msg.task_id, response["output"])
    
    await hub.connect()
    await hub.run_forever()

Environment Variables

• GOPHERHOLE_API_KEY - Your API key (used by GopherHole.from_env())

License

MIT