How to Add a New Platform Channel to Agent Reach: Complete Channel Contract Guide

Adding a new platform to Agent Reach requires creating a Python class that inherits from Channel in agent_reach/channels/base.py and implements the four-method contract—can_handle, read, search, and check—alongside the required metadata attributes name, description, backends, and tier.

Agent Reach is an open-source automation framework that unifies interactions across internet platforms through a standardized channel abstraction. To extend its capabilities to a new service—whether a social network, code repository, or content platform—you must implement the Agent Reach channel contract defined in the base class. This guide walks through the architectural requirements and concrete implementation steps using the actual source code from the Panniantong/Agent-Reach repository.

Understanding the Agent Reach Channel Contract

The contract is defined by the abstract Channel class in agent_reach/channels/base.py. Every concrete channel must provide specific metadata attributes and implement four core methods that enable the routing engine to dispatch URLs and queries appropriately.

Required Class Attributes

• name (str): The short identifier used in configuration files and CLI commands (e.g., "twitter", "github").
• description (str): Human-readable summary displayed in help text and documentation.
• backends (List[str]): Priority-ordered list of supported backends (CLI tool names, API identifiers, or service wrappers).
• tier (int): Configuration complexity level—0 for zero-config, 1 for requiring free API keys, 2 for full enterprise setup.

Required Implementation Methods

• can_handle(self, url: str) -> bool: Determines if the channel can process a given URL by inspecting the domain or path structure.
• read(self, url: str) -> str: Retrieves content from a specific URL using the active_backend selected during initialization.
• search(self, query: str) -> str: Executes platform-specific search queries and returns formatted results.
• check(self, config=None) -> Tuple[str, str]: Probes each backend in ordered_backends() to verify availability, sets self.active_backend to the first working option, and returns a status tuple (status, message) where status is "ok", "warn", or "error".

Step-by-Step Implementation Guide

Step 1 – Create the Channel Module

Create a new file at agent_reach/channels/<platform>.py using snake_case naming consistent with your channel's name attribute.

Step 2 – Implement the Channel Class

Subclass Channel and define the metadata attributes. Import the base class and probing utilities:

from .base import Channel
from agent_reach.probe import probe_command
import subprocess

Define the class skeleton:

class NewPlatformChannel(Channel):
    name = "newplatform"
    description = "New Platform integration for Agent Reach"
    backends = ["new-cli", "new-api"]
    tier = 1

Step 3 – Implement URL Detection (can_handle)

The can_handle method must parse the URL and return True for domains belonging to your platform:

def can_handle(self, url: str) -> bool:
    from urllib.parse import urlparse
    netloc = urlparse(url).netloc.lower()
    return "newplatform.com" in netloc

Step 4 – Implement Content Retrieval (read)

Delegate to the active backend to fetch content. The active_backend is set by the check method during channel initialization:

def read(self, url: str) -> str:
    cmd = [self.active_backend, "read", url, "--output", "yaml"]
    result = subprocess.run(cmd, capture_output=True, text=True)
    return result.stdout

Step 5 – Implement Search Functionality (search)

Similar to read, but accepting free-form query strings:

def search(self, query: str) -> str:
    cmd = [self.active_backend, "search", query, "--output", "yaml"]
    result = subprocess.run(cmd, capture_output=True, text=True)
    return result.stdout

Step 6 – Implement Backend Health Checks (check)

Following the pattern in agent_reach/channels/twitter.py (lines 29-48), iterate through ordered_backends() and probe each candidate:

def check(self, config=None):
    self.active_backend = None
    findings = []

    for backend in self.ordered_backends(config):
        if backend == "new-cli":
            probe = probe_command("new-cli", ["status"], timeout=15, package="new-cli")
            if probe.status == "missing":
                continue
            if probe.ok:
                self.active_backend = backend
                return "ok", "new-cli fully operational"
            findings.append((backend, "error" if probe.status == "broken" else "warn", 
                           "new-cli installed but authentication failed"))
    
    # Fallback logic

    for status in ["ok", "warn", "error"]:
        for backend, stat, msg in findings:
            if stat == status:
                self.active_backend = backend
                return stat, msg
    return "warn", "No backends available"

Step 7 – Register the Channel

Expose the implementation by editing agent_reach/channels/__init__.py to import the new class:

from .newplatform import NewPlatformChannel

If your version uses an explicit registry in agent_reach/core.py, add the channel to the CHANNELS dictionary:

CHANNELS = {
    "twitter": TwitterChannel(),
    "github": GitHubChannel(),
    "newplatform": NewPlatformChannel(),  # Add this line

}

Complete Working Example

Here is a full implementation skeleton for a hypothetical platform:

# agent_reach/channels/newplatform.py

"""NewPlatform channel implementation for Agent Reach."""

from .base import Channel
from agent_reach.probe import probe_command
import subprocess


class NewPlatformChannel(Channel):
    name = "newplatform"
    description = "New Platform – read posts and search content"
    backends = ["new-cli", "new-api"]
    tier = 1

    def can_handle(self, url: str) -> bool:
        from urllib.parse import urlparse
        return "newplatform.com" in urlparse(url).netloc.lower()

    def read(self, url: str) -> str:
        cmd = [self.active_backend, "read", url, "--output", "yaml"]
        result = subprocess.run(cmd, capture_output=True, text=True)
        return result.stdout

    def search(self, query: str) -> str:
        cmd = [self.active_backend, "search", query, "--output", "yaml"]
        result = subprocess.run(cmd, capture_output=True, text=True)
        return result.stdout

    def check(self, config=None):
        self.active_backend = None
        findings = []

        for backend in self.ordered_backends(config):
            if backend == "new-cli":
                probe = probe_command("new-cli", ["status"], timeout=15, package="new-cli")
                if probe.status == "missing":
                    continue
                if probe.ok:
                    self.active_backend = backend
                    return "ok", "new-cli fully available"
                findings.append((backend, "error", "new-cli broken"))
            elif backend == "new-api":
                # API key validation logic here

                pass

        for backend, status, msg in findings:
            self.active_backend = backend
            return status, msg
        return "warn", "No backends configured"

Key Source Files Reference

• agent_reach/channels/base.py: Defines the abstract Channel class and ordered_backends() helper method.
• agent_reach/channels/twitter.py: Reference implementation demonstrating multi-backend probing (see lines 29-48 for check method patterns).
• agent_reach/channels/github.py: Minimal implementation example for single-backend channels.
• agent_reach/channels/__init__.py: Module exports that make channels discoverable by the core router.
• agent_reach/core.py: Central dispatcher that imports channels and constructs the routing registry.
• agent_reach/probe.py: Utility module providing probe_command() for backend health verification.

Summary

• The Agent Reach channel contract requires implementing four methods—can_handle, read, search, and check—plus four metadata attributes (name, description, backends, tier).
• Inherit from Channel in agent_reach/channels/base.py to gain access to ordered_backends() and standard initialization logic.
• Place new channel implementations in agent_reach/channels/<platform>.py and expose them via agent_reach/channels/__init__.py.
• Use probe_command from agent_reach.probe to validate CLI backends within your check method, following the pattern established in agent_reach/channels/twitter.py.
• Set self.active_backend during check() to ensure read() and search() have a valid target for subprocess calls.

Frequently Asked Questions

What happens if I don't implement the check method?

If you omit check, the base class provides a default implementation, but you must still set self.active_backend manually or the channel will fail at runtime when read() or search() attempts to access it. The check method is the recommended location to probe backends and establish connectivity before operations begin.

Can I support multiple backends for failover?

Yes. Populate the backends list with ordered priorities (e.g., ["premium-api", "free-cli", "legacy-tool"]). The ordered_backends() method from the base class respects user configuration overrides while maintaining your declared precedence. Iterate through this list in check() to select the first available option, as demonstrated in agent_reach/channels/twitter.py.

How does the routing engine know which channel handles a URL?

The core dispatcher in agent_reach/core.py calls can_handle(url) on every registered channel until one returns True. Implement precise domain matching or path patterns in this method to ensure Agent Reach routes URLs to your channel correctly without false positives that could intercept traffic intended for other platforms.

Where should I store API keys or configuration for my channel?

Store sensitive configuration in the user-level Agent Reach config file. Access these values via the config parameter passed to check(config). The base class handles config loading, making platform-specific settings available as dictionaries keyed by your channel's name attribute, allowing you to validate credentials during the health check phase.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}