Hook Reference¶

Pynchy plugins implement hooks defined in src/pynchy/plugins/hookspecs.py. Each hook corresponds to a plugin category, and a plugin can implement any combination of hooks.

All hooks use pluggy's @hookimpl decorator:

import pluggy

hookimpl = pluggy.HookimplMarker("pynchy")

pynchy_agent_core_info¶

Provide an alternative LLM agent framework.

Calling strategy: All results collected (multiple cores can coexist; selected via PYNCHY_AGENT_CORE env var).

@hookimpl
def pynchy_agent_core_info(self) -> dict[str, str | list[str] | None]:
    return {
        "name": "ollama",                              # Core identifier
        "module": "pynchy_plugin_ollama.core",         # Python module path
        "class_name": "OllamaAgentCore",               # Class to instantiate
        "packages": ["ollama>=0.1.0"],                 # pip packages for container
        "host_source_path": str(Path(__file__).parent), # Source to mount, or None
    }

Return keys:

pynchy_service_handler¶

Provide host-side handlers for service tools. This hook provides handler functions that run on the host process and are dispatched to via IPC when container agents invoke service tools.

Calling strategy: All results collected; tool mappings are merged (last-write-wins on conflict).

@hookimpl
def pynchy_service_handler(self) -> dict[str, Any]:
    return {
        "tools": {
            "list_calendar": _handle_list_calendar,
            "create_event": _handle_create_event,
            "delete_event": _handle_delete_event,
        },
    }

Return keys:

Handler function signature:

async def handler(data: dict) -> dict:
    """Process a service tool request.

    Args:
        data: The full IPC request dict (includes type, request_id, and tool-specific fields)

    Returns:
        Dict with either {"result": ...} on success or {"error": "..."} on failure
    """

Request flow: Container MCP tool → IPC request → host policy check → plugin handler → IPC response

pynchy_skill_paths¶

Provide agent skills (markdown instruction files) that get mounted into the container.

Calling strategy: All results collected and flattened. Skills are filtered per-workspace based on the skills config field before being copied into the session directory.

@hookimpl
def pynchy_skill_paths(self) -> list[str]:
    return [str(Path(__file__).parent / "skills" / "code-review")]

Return value: List of absolute paths to skill directories. Each directory should contain a SKILL.md file following the Claude Agent SDK skill format.

Skill directory structure:

skills/
└── code-review/
    ├── SKILL.md          # Required: skill definition
    └── examples.md       # Optional: supporting files

SKILL.md frontmatter:

Skills declare metadata via YAML frontmatter at the top of SKILL.md:

---
name: code-review
description: Review code for bugs and style issues.
tier: community
---

Skill tiers:

Workspaces opt into skills via the skills config field:

[workspaces.my-workspace]
skills = ["core", "dev"]           # tier names and/or individual skill names

When skills is not set, only core-tier skills are included (safe default). When set, entries are unioned — ["core", "my-skill"] means all core-tier skills plus my-skill specifically. Core is always implicit when any filtering is active. Use ["all"] to include every available skill.

pynchy_create_channel¶

Provide a communication channel (Telegram, Slack, Discord, etc.).

Calling strategy: All non-None channels are collected; host config chooses the default channel.

@hookimpl
def pynchy_create_channel(self, context: Any) -> Any | None:
    bot_token = os.environ.get("TELEGRAM_BOT_TOKEN")
    if not bot_token:
        return None  # This plugin doesn't apply
    return TelegramChannel(
        bot_token=bot_token,
        on_message=context.on_message_callback,
        on_chat_metadata=context.on_chat_metadata_callback,
        registered_groups=context.registered_groups,
    )

Parameters:

ChannelPluginContext fields:

Return value: A Channel instance implementing the channel protocol, or None to pass.

Channel protocol:

class Channel(Protocol):
    name: str

    async def connect(self) -> None: ...
    async def send_message(self, jid: str, text: str) -> None: ...
    def is_connected(self) -> bool: ...
    def owns_jid(self, jid: str) -> bool: ...
    async def disconnect(self) -> None: ...

Optional attributes (check with hasattr/getattr): prefix_assistant_name (bool, default True), set_typing, create_group.

pynchy_container_runtime¶

Provide a host container runtime implementation (for example Apple Container).

Calling strategy: All results collected; runtime selection picks by config override ([container].runtime) or platform-aware auto-detection.

@hookimpl
def pynchy_container_runtime(self) -> Any | None:
    return AppleContainerRuntime()

Runtime object contract:

pynchy_tunnel¶

Provide a tunnel provider for remote connectivity detection (Tailscale, Cloudflare Tunnel, WireGuard, etc.).

Calling strategy: All results collected; pynchy checks each provider at startup and warns if none are connected.

@hookimpl
def pynchy_tunnel(self) -> Any | None:
    return MyTunnelProvider()

Tunnel provider contract:

Built-in: Tailscale is provided as a built-in plugin (src/pynchy/tunnels/plugins/tailscale.py). It shells out to tailscale status --json and checks BackendState.

pynchy_observer¶

Provide an event observer that subscribes to the EventBus and persists or processes events (SQLite, OpenTelemetry, log files, etc.).

Calling strategy: All results collected; each observer's subscribe() is called with the event bus during startup.

@hookimpl
def pynchy_observer(self) -> Any | None:
    return SqliteEventObserver()

Observer object contract:

Event types available:

Built-in: The SQLite observer (src/pynchy/observers/plugins/sqlite_observer/) stores all events to a dedicated events table in the main database.

pynchy_memory¶

Provide a persistent memory backend (save, recall, forget, list). Agents use memory tools to store facts across sessions.

Calling strategy: All results collected; first non-None result wins.

@hookimpl
def pynchy_memory(self) -> Any | None:
    return MyMemoryBackend()

Memory backend contract:

Parameters:

Built-in: The SQLite memory plugin (src/pynchy/memory/plugins/sqlite_memory/) stores memories in the main database with FTS5 full-text search.

pynchy_mcp_server_spec¶

Provide an MCP server specification. Plugin-provided specs are merged with user-defined servers in config.toml. Config.toml definitions override plugin defaults when both use the same server name.

Calling strategy: All results collected and merged. A plugin can return a single dict or a list of dicts (for plugins providing multiple servers).

@hookimpl
def pynchy_mcp_server_spec(self) -> list[dict[str, Any]]:
    return [
        {
            "name": "gdrive",
            "type": "docker",
            "image": "pynchy-mcp-gdrive:latest",
            "dockerfile": "src/pynchy/agent/mcp/gdrive.Dockerfile",
            "port": 3100,
            "transport": "streamable_http",
            "env": {"GDRIVE_OAUTH_PATH": "/home/chrome/gcp-oauth.keys.json"},
        },
    ]

Return keys:

Instance expansion: Users don't configure the base spec — they declare instances in config.toml that reference the plugin-provided template:

[mcp_servers.gdrive.anyscale]
chrome_profile = "anyscale"

The MCP manager merges this with the plugin-provided base spec, auto-assigns ports, and mounts chrome profile directories. See MCP Servers for user-facing config details.

pynchy_workspace_spec¶

Provide a managed workspace definition (for example a periodic agent).

Calling strategy: All results collected and merged with user config.toml workspaces.

@hookimpl
def pynchy_workspace_spec(self) -> dict[str, Any]:
    return {
        "folder": "code-improver",
        "config": {
            "pynchy_repo_access": True,
            "schedule": "0 4 * * *",
            "prompt": "Run scheduled code improvements",
            "context_mode": "isolated",
        },
    }

Return keys:

Agent instructions are now delivered via directives rather than seeded CLAUDE.md files. The claude_md field is ignored.

Multi-Category Plugins¶

A single plugin can implement multiple hooks:

class CalendarPlugin:
    """Provides calendar service handlers AND calendar skills."""

    @hookimpl
    def pynchy_service_handler(self) -> dict:
        return {
            "tools": {
                "list_calendar": _handle_list_calendar,
                "create_event": _handle_create_event,
            },
        }

    @hookimpl
    def pynchy_skill_paths(self) -> list[str]:
        return [str(Path(__file__).parent / "skills" / "calendar")]

No categories attribute needed — pluggy determines capabilities from which hooks the class implements.

Hook Execution Order¶

Pluggy supports ordering hints:

@hookimpl(trylast=True)   # Run after other plugins
@hookimpl(tryfirst=True)  # Run before other plugins

Most plugins don't need this. Use it when one plugin needs to see or modify another plugin's results.