Skip to content

agent-context-graph: entry_points-based runtime plugin registry #273

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
antejavor merged 1 commit into from
Aug 12, 2026
Merged

agent-context-graph: entry_points-based runtime plugin registry #273

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 43 additions & 20 deletions context-graph/agent-context-graph/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -406,29 +406,30 @@ class MyRuntimeAdapter(RuntimeAdapter):

### Adding a New Command-Hook Runtime Adapter

The pattern above fits **in-process** runtimes — something that calls your Python code directly (an SDK callback, an embedded framework). **Command-hook** runtimes are different: the harness invokes an external command with a JSON payload on stdin (Claude Code, Codex), rather than calling into your process. That needs three extra pieces beyond `RuntimeAdapter` itself, all shown in full in `adapters/claude_code.py` and `adapters/codex.py`:
The pattern above fits **in-process** runtimes — something that calls your Python code directly (an SDK callback, an embedded framework). **Command-hook** runtimes are different: the harness invokes an external command with a JSON payload on stdin (Claude Code, Codex), rather than calling into your process.

1. **A payload translator.** Same idea as `RuntimeAdapter.get_runtime_hooks()`, but the input is the harness's raw JSON payload (read from stdin) rather than a native callback. Map each of the harness's hook event names to the matching `Event` subclass and call `link.emit(...)` — see `ClaudeCodeHooksAdapter._events_from_payload` for the full mapping.
2. **A hook-config generator.** A plain function that builds whatever config shape the harness expects for wiring hooks (a `hooks.json`-style block, a plugin manifest, ...), pointing every hook at your CLI entry point. See `build_hooks_config` in `adapters/claude_code.py`.
3. **A CLI entry point.** Reads one JSON payload from stdin (`load_payload`), constructs an `AgentLink` with the requested connectors, feeds the payload through your adapter, and — if the harness expects one — writes a JSON response to stdout (`response_for_payload`). This is what the generated hook config actually invokes.
Three pieces beyond `RuntimeAdapter` itself, exactly as `adapters/claude_code.py` and `adapters/codex.py` implement them:

Skeleton:
1. **A payload translator.** Same idea as `RuntimeAdapter.get_runtime_hooks()`, but the input is the harness's raw JSON payload rather than a native callback. Map each of the harness's hook event names to the matching `Event` subclass and call `link.emit(...)`.
2. **A hook-config generator** (`build_hooks_config(command)`). Builds whatever config shape the harness expects for wiring hooks, pointing every hook at your CLI entry point.
3. **A response function** (`response_for_payload(payload)`). Returns the JSON the harness expects back on stdout, or `None`.

The stdin-loading, connector-construction, and CLI-argument-parsing scaffolding around those three pieces is **shared** — `hooks/runner.py`'s `run_hook(plugin, argv)` does that for every registered runtime, so a new adapter doesn't write its own `main()` at all. Register your runtime as a **plugin** and the generic runner (plus `bootstrap`/`doctor`/`hook run`/`hook init`) picks it up automatically — no changes to `agent-context-graph` itself:

```python
import json
import sys
# my_package/adapter.py
from dataclasses import dataclass

from agent_context_graph.link import AgentLink
from agent_context_graph.protocols import RuntimeAdapter


class MyCommandHookAdapter(RuntimeAdapter):
def __init__(self, link: AgentLink, session_id: str | None = None):
def __init__(self, link, session_id: str | None = None):
self._link = link
self._session_id = session_id

def get_runtime_hooks(self):
return build_hooks_config("my-adapter hook run")
return build_hooks_config("my-runtime hook run my-runtime")

def handle_payload(self, payload: dict) -> None:
for event in self._events_from_payload(payload):
Expand All @@ -439,21 +440,43 @@ class MyCommandHookAdapter(RuntimeAdapter):
...


def build_hooks_config(command: str) -> dict:
# Return whatever config format your harness expects, every hook
# pointing at `command`.
def build_hooks_config(command: str, *, timeout: int = 30) -> dict:
# Return whatever config format your harness expects, every hook pointing at `command`.
...


def response_for_payload(payload: dict) -> dict | None:
# Return the JSON your harness expects back, or None.
...


def main() -> int:
payload = json.loads(sys.stdin.read() or "{}")
link = AgentLink()
# ... add_connector(...) per --connector flags, as in claude_code.py's create_link
MyCommandHookAdapter(link).handle_payload(payload)
return 0
@dataclass(frozen=True)
class MyRuntimePlugin:
name: str = "my-runtime"
adapter_class: type = MyCommandHookAdapter

def response_for_payload(self, payload: dict) -> dict | None:
return response_for_payload(payload)

def build_hooks_config(self, command: str, *, timeout: int = 30) -> dict:
return build_hooks_config(command, timeout=timeout)

# init(project_dir, connectors, **kwargs) is optional -- omit it if your
# runtime has no project-local hook-config file to generate (matching
# ClaudeCodeHooksAdapter's own plugin, which doesn't define one yet).


PLUGIN = MyRuntimePlugin()
```

Then register it in your own package's `pyproject.toml` — this is the entire integration, no fork or PR against this repo required:

```toml
[project.entry-points."agent_context_graph.runtimes"]
my-runtime = "my_package.adapter:PLUGIN"
```

Note: `claude_code.py` and `codex.py` currently duplicate the stdin-loading/connector-construction/CLI-runner scaffolding rather than sharing a helper module for it (flagged in a `# TODO` in `codex.py`) — a third adapter is a good trigger to finally extract that shared plumbing, but isn't required to add one today.
Once installed, `agent-context-graph bootstrap --runtime my-runtime`, `doctor --runtime my-runtime`, `hook run my-runtime`, and `hook init my-runtime` (if `init` is implemented) all work exactly like the built-in Codex and Claude Code plugins — see `runtime_plugin.py` for the full protocol and `pyproject.toml`'s own `[project.entry-points."agent_context_graph.runtimes"]` for how Codex/Claude Code register themselves.

### Adding a New Graph Component

Expand Down
4 changes: 4 additions & 0 deletions context-graph/agent-context-graph/pyproject.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,10 @@ dependencies = [
[project.scripts]
agent-context-graph = "agent_context_graph.cli:main"

[project.entry-points."agent_context_graph.runtimes"]
codex = "agent_context_graph.adapters.codex:PLUGIN"
claude-code = "agent_context_graph.adapters.claude_code:PLUGIN"

[project.optional-dependencies]
claude = [
"claude-agent-sdk>=0.1.0",
Expand Down
206 changes: 29 additions & 177 deletions context-graph/agent-context-graph/src/agent_context_graph/adapters/claude_code.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,10 +7,7 @@

from __future__ import annotations

import argparse
import json
import os
import sys
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any

from agent_context_graph.events import (
Expand All @@ -23,13 +20,14 @@
ToolEndEvent,
ToolStartEvent,
)
from agent_context_graph.link import AgentLink
from agent_context_graph.hooks.runner import create_link, load_payload # noqa: F401 (re-exported for callers)
from agent_context_graph.protocols import RuntimeAdapter

if TYPE_CHECKING:
from collections.abc import Iterable, Sequence
from collections.abc import Sequence

from agent_context_graph.events import Event
from agent_context_graph.link import AgentLink

_SOURCE = "claude-code"
_DEFAULT_COMMAND = "agent-context-graph hook run claude-code"
Expand Down Expand Up @@ -239,46 +237,6 @@ def build_hooks_config(command: str, *, timeout: int = 30) -> dict[str, list[dic
return config


def load_payload(stream: Any | None = None) -> dict[str, Any]:
"""Read one Claude Code hook payload from a text stream."""
if stream is None:
stream = sys.stdin
raw = stream.read()
if not raw.strip():
return {}
payload = json.loads(raw)
if not isinstance(payload, dict):
msg = "Claude Code hook payload must be a JSON object"
raise TypeError(msg)
return payload


def create_link(connector_names: Iterable[str] = (), *, memgraph_env: dict[str, str] | None = None) -> AgentLink:
"""Create an AgentLink with optional connectors named by CLI/config.

Args:
connector_names: Which graph connectors to attach.
memgraph_env: Resolved Memgraph connection dict (keys: MEMGRAPH_URL,
MEMGRAPH_USER, MEMGRAPH_PASSWORD, MEMGRAPH_DATABASE). When None,
connectors fall back to their own default resolution.
"""
link = AgentLink()
for connector_name in connector_names:
normalized = connector_name.strip().replace("-", "_")
if not normalized:
continue
if normalized == "skills_graph":
_add_skills_graph_connector(link, memgraph_env)
elif normalized == "actions_graph":
_add_actions_graph_connector(link, memgraph_env)
elif normalized == "sessions_graph":
_add_sessions_graph_connector(link, memgraph_env)
else:
msg = f"Unsupported connector: {connector_name}"
raise ValueError(msg)
return link


def response_for_payload(payload: dict[str, Any]) -> dict[str, Any] | None:
"""Return hook JSON response, when Claude Code benefits from one."""
hook_event_name = payload.get("hook_event_name")
Expand All @@ -287,133 +245,32 @@ def response_for_payload(payload: dict[str, Any]) -> dict[str, Any] | None:
return None


@dataclass(frozen=True)
class _ClaudeCodePlugin:
"""Registered under the ``agent_context_graph.runtimes`` entry point as ``PLUGIN``.

No ``init`` -- Claude Code project-local hook setup isn't implemented yet
(see ``hooks/cli.py``'s generic ``_init`` dispatch, which reports that
clearly rather than assuming every runtime supports it).
"""

name: str = "claude-code"
adapter_class: type[RuntimeAdapter] = ClaudeCodeHooksAdapter

def response_for_payload(self, payload: dict[str, Any]) -> dict[str, Any] | None:
return response_for_payload(payload)

def build_hooks_config(self, command: str, *, timeout: int = 30) -> dict[str, Any]:
return build_hooks_config(command, timeout=timeout)


PLUGIN = _ClaudeCodePlugin()


def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Bridge Claude Code hooks to agent-context-graph.")
parser.add_argument(
"--connector",
action="append",
default=None,
help="Graph connector to enable. Currently supported: skills-graph, actions-graph, sessions-graph.",
)
parser.add_argument(
"--session-id",
default=None,
help="Override the session id from the Claude Code hook payload.",
)
parser.add_argument(
"--memgraph-url",
default=None,
help="Memgraph Bolt URL. Overrides config file value.",
)
parser.add_argument(
"--memgraph-user",
default=None,
help="Memgraph username. Overrides config file value.",
)
parser.add_argument(
"--memgraph-password",
default=None,
help="Memgraph password. Overrides config file value.",
)
parser.add_argument(
"--memgraph-database",
default=None,
help="Memgraph database. Overrides config file value.",
)
parser.add_argument(
"--strict",
action="store_true",
help="Return a non-zero status if the hook payload cannot be recorded.",
)
args = parser.parse_args(argv)

connector_names = args.connector
if connector_names is None:
connector_names = _connectors_from_env()

# Resolve Memgraph connection: CLI flag > config file > default.
from agent_context_graph.adapters._identity import resolve_memgraph_env

memgraph_env = resolve_memgraph_env(
url=args.memgraph_url,
user=args.memgraph_user,
password=args.memgraph_password,
database=args.memgraph_database,
)

payload: dict[str, Any] = {}
try:
payload = load_payload()
link = create_link(connector_names, memgraph_env=memgraph_env)
adapter = ClaudeCodeHooksAdapter(link, session_id=args.session_id)
adapter.handle_payload(payload)
response = response_for_payload(payload)
if response is not None:
print(json.dumps(response))
except Exception as exc:
if args.strict or os.environ.get("AGENT_CONTEXT_GRAPH_CLAUDE_CODE_STRICT") == "1":
raise
response = response_for_payload(payload)
if response is not None:
print(json.dumps(response))
_debug_log(f"agent-context-graph Claude Code hook skipped: {exc}")
return 0


def _add_skills_graph_connector(link: AgentLink, memgraph_env: dict[str, str] | None = None) -> None:
try:
from skills_graph import SkillGraph
from skills_graph.connector import SkillGraphConnector
except ImportError as exc:
msg = "skills-graph is required for the skills-graph Claude Code connector"
raise ImportError(msg) from exc

kwargs = _memgraph_kwargs(memgraph_env)
graph = SkillGraph(**kwargs)
link.add_connector(SkillGraphConnector(graph))


def _add_sessions_graph_connector(link: AgentLink, memgraph_env: dict[str, str] | None = None) -> None:
try:
from sessions_graph import SessionsGraph
from sessions_graph.connector import SessionsGraphConnector
except ImportError as exc:
msg = "sessions-graph is required for the sessions-graph Claude Code connector"
raise ImportError(msg) from exc

kwargs = _memgraph_kwargs(memgraph_env)
graph = SessionsGraph(**kwargs)
link.add_connector(SessionsGraphConnector(graph))


def _add_actions_graph_connector(link: AgentLink, memgraph_env: dict[str, str] | None = None) -> None:
try:
from actions_graph import ActionsGraph
from actions_graph.connector import ActionsGraphConnector
except ImportError as exc:
msg = "actions-graph is required for the actions-graph Claude Code connector"
raise ImportError(msg) from exc

kwargs = _memgraph_kwargs(memgraph_env)
graph = ActionsGraph(**kwargs)
link.add_connector(ActionsGraphConnector(graph))


def _memgraph_kwargs(memgraph_env: dict[str, str] | None) -> dict[str, str]:
"""Convert resolved memgraph env dict to kwargs for graph component constructors."""
if memgraph_env is None:
return {}
return {
"url": memgraph_env["MEMGRAPH_URL"],
"username": memgraph_env["MEMGRAPH_USER"],
"password": memgraph_env["MEMGRAPH_PASSWORD"],
"database": memgraph_env["MEMGRAPH_DATABASE"],
}


def _connectors_from_env() -> list[str]:
value = os.environ.get("AGENT_CONTEXT_GRAPH_CLAUDE_CODE_CONNECTORS", "")
return [part.strip() for part in value.split(",") if part.strip()]
from agent_context_graph.hooks.runner import run_hook

return run_hook(PLUGIN, argv)


def _metadata_from_payload(payload: dict[str, Any]) -> dict[str, Any]:
Expand Down Expand Up @@ -464,10 +321,5 @@ def _resolve_user_id(payload: dict[str, Any]) -> str | None:
return resolve_user_id(payload)


def _debug_log(message: str) -> None:
if os.environ.get("AGENT_CONTEXT_GRAPH_CLAUDE_CODE_DEBUG") == "1":
print(message, file=sys.stderr)


if __name__ == "__main__":
raise SystemExit(main())
Loading
Loading