# mcp-core

Shared MCP-server machinery for the DNS platform Flask apps (SR, Books, CRM,
DM, PM). Each app currently hand-rolls the same five things in its
`app/routes/mcp.py`; this library extracts them once, behind interfaces:

| Module | What it extracts | Reference implementation |
| --- | --- | --- |
| `mcp_core.transport` | Streamable HTTP transport (spec rev 2025-03-26): single `POST /mcp` endpoint, JSON-RPC 2.0 + batches, stateless (`GET` -> 405, `DELETE` -> 200), bearer required on every request | SR `mcp_streamable` |
| `mcp_core.auth` | `TokenValidator` protocol + sha256 token-hash helpers (no plaintext-recoverable storage, ever) | SR/Books `validate_bearer_token` |
| `mcp_core.identity` | `CallerResolver` / `PrincipalResolver` protocols, fail-closed caller resolution, explicit `with_external_tiers` opt-in | SR `resolve_caller`, Books `_resolve_service_caller` |
| `mcp_core.registry` | `ToolRegistry`: declarative per-tool permissions, central identity + permission gate, dynamic per-call re-gating, per-caller `tools/list`, caller-identity schema injection | Books `TOOL_PERMISSIONS` + `execute_tool` gate, SR `_tools_for_caller` |
| `mcp_core.errors` | The **frozen** tool-error contract (`{"error": ...}` inside a successful JSON-RPC result) | both |
| `mcp_core.audit` | `AuditSink` protocol + logging sink; one event per dispatch with (org, caller, tool, outcome) | new (was implicit log lines) |
| `mcp_core.authz_ui` | Catalog-driven Members + Roles admin pages (0.2.0): `PermissionCatalog` + `AuthzProvider` protocol + blueprint factory rendering the roles matrix from the catalog | Books `PERMISSION_LABELS` + settings Users & Permissions UI |

App-agnostic by construction: **zero imports from any app**, no SQLAlchemy
dependency, no models. Every integration point is a `typing.Protocol` the
app implements over its own tables. Flask is the only runtime dependency
(the transport is a Blueprint); the app decides the Flask/SQLAlchemy
versions in play.

## Frozen contracts

These are wire contracts the Phone app, ChatGPT connectors, and cross-app
service-token callers already depend on. Changing any of them requires a
coordinated migration across every consumer — treat them as frozen:

1. **Tool errors are successful JSON-RPC results.** A refused or failed
   tool call returns `"result"` whose `content[0].text` is the JSON of
   `{"error": "<message>"}`. JSON-RPC `"error"` objects are protocol-level
   only (-32700 parse, -32601 unknown method, -32000 unauthenticated,
   -32603 unexpected handler exception).
2. **Frozen error strings** (agents parse and act on these):
   - `Unknown tool: {name}`
   - `caller_phone or caller_email is required when calling via a service token`
   - `Permission denied: {permission}`
3. **Result serialization**: the tool result dict is serialized with
   `json.dumps(result, indent=2)` into a single `text` content block.
4. **Endpoint semantics**: `POST <prefix>` (and `<prefix>/`) for JSON-RPC;
   `GET` -> 405 with `Allow: POST, DELETE`; `DELETE` -> 200 no-op; missing
   or invalid bearer -> 401 with JSON-RPC error code -32000.

## Fail-closed guarantees (preserved from the shipped slices)

- **Unknown tool name** -> refused before anything else is inspected.
- **Service token without `caller_phone`/`caller_email`** -> refused, unless
  the tool is registered `identity_exempt=True`. Keep that set tiny and
  structural (machine-to-machine callbacks only — SR's `receive_call_result`
  is the canonical member). Never exempt a tool a human invokes.
- **Unresolvable caller** (unknown identity, misconfigured staff role, no
  org membership, no resolver configured) -> refused. Resolvers signal this
  by raising `IdentityError`; the message is surfaced verbatim as the tool
  error.
- **Unmapped permission** -> a tool declares its requirement at
  registration; `dynamic_permission` deriving `None` from the arguments
  refuses the call. There is no allow-by-default path anywhere.
- **No fallback tiers.** An unknown caller never becomes a guest. SR-style
  anonymous external tiers are an explicit opt-in via
  `with_external_tiers(staff_resolver, classifier)` — and even then the
  fallback is phone-keyed only; an unknown `caller_email` always fails
  closed (an email must never bootstrap an identity).

## Usage sketch

```python
from mcp_core import (
    AnyOf, Caller, IdentityError, McpServer, TokenContext, ToolRegistry,
    hash_token, streamable_http_blueprint,
)

# 1) Bearer validation over the app's oauth_tokens table.
class AppTokenValidator:
    def validate(self, bearer):
        row = OAuthToken.query.filter_by(access_token_hash=hash_token(bearer)).first()
        if not row or not row.is_access_valid():
            return None
        if row.organization_id:
            return TokenContext(org_id=str(row.organization_id), kind="service")
        user = User.query.get(row.user_id)
        if not user or not user.is_active():
            return None
        return TokenContext(org_id=None, kind="user", principal_user_id=str(user.id))

# 2) Caller resolution over the app's users/memberships.
class AppCallerResolver:
    def resolve(self, org_id, caller_phone=None, caller_email=None):
        ou = ...  # phone wins over email; search ONLY org_id; normalize app-side
        if ou is None:
            raise IdentityError("caller not recognized in this organization")
        return Caller(user_id=str(ou.user_id), role=ou.role,
                      permissions=frozenset(perms_for(ou)), extra={"org_user": ou})

class AppPrincipalResolver:
    def resolve_principal(self, token, org_id):
        ou = ...  # verify ACTIVE membership of token.principal_user_id in org_id
        if ou is None:
            raise IdentityError("Access denied to this organization")
        return Caller(...)

# 3) Declare tools with the app's permission keys.
registry = ToolRegistry(
    caller_resolver=AppCallerResolver(),
    principal_resolver=AppPrincipalResolver(),
    permission_evaluator=lambda caller, key: ...,  # optional: has_permission semantics
)

@registry.tool("list_invoices", permission="ar.invoices.view",
               schema={"type": "object", "properties": {...}})
def list_invoices(ctx, args):
    # ctx.caller is the resolved acting person; ctx.org_id the tenant.
    return {"invoices": [...]}

# 4) Mount the transport.
server = McpServer(registry=registry, validator=AppTokenValidator(),
                   name="Books MCP Server", version="1.0.0")
app.register_blueprint(streamable_http_blueprint(server))  # serves POST /mcp
```

### 0.1.1 additions (SR pilot)

- `ToolRegistry(visibility_filter=...)`: optional app-owned `tools/list`
  view — `(token, org_id) -> iterable of visible tool names | None`.
  Returning names advertises exactly those registered tools (registration
  order); `None` falls through to the default logic. Advertisement-only;
  never gates dispatch. Built for apps that migrate with `permission=None`
  registrations (per-handler gates kept in place, e.g. SR): their
  per-caller listing can't be derived from declared requirements.
- `mcp_core.transport.dispatch_message(server, token, data)` is now public:
  legacy transports still mounted during a migration (SR's `/mcp/sse` POST)
  route their JSON-RPC through the same single dispatch core. The caller
  owns bearer validation and the HTTP envelope.

## 0.2.0 additions — `mcp_core.authz_ui` (catalog-driven admin UI)

New subpackage: shared Members + Roles admin pages generated from a
per-app **permission catalog**, so all five apps get one consistent roles
matrix instead of five hand-built ones. Same construction rules as the
rest of the library — no database access, no app imports, every
integration point a protocol — but it is admin-UI machinery, not
MCP-server machinery, so it is *not* re-exported from the top-level
`mcp_core` namespace. Import from `mcp_core.authz_ui` directly.

### 1) Declare a catalog

The catalog is the single source of truth: ordered groups (module ->
entries), each entry a dotted key (`module.entity.action[.scope]`, 2–4
lowercase segments) plus display label and optional description. Keys are
the SAME strings the app's web gates and MCP tool registrations use.
Validation (well-formed keys, uniqueness, non-empty labels/groups) runs at
construction, so a bad catalog fails at app startup, never in front of an
admin.

Books adopts its existing `PERMISSION_LABELS` mapping verbatim:

```python
from mcp_core.authz_ui import PermissionCatalog
from app.utils.permissions import PERMISSION_LABELS, PERMISSION_INHERITANCE

catalog = PermissionCatalog.from_dict(
    PERMISSION_LABELS, inheritance=PERMISSION_INHERITANCE
)
```

Apps without a labels dict declare groups directly:

```python
catalog = PermissionCatalog(
    [
        ("Properties", [
            ("pm.properties.view", "View properties"),
            ("pm.properties.manage", "Manage properties",
             "Create, edit, archive properties"),
        ]),
        ("Users & Permissions", [
            ("settings.users.view", "View team members"),
            ("settings.users.invite", "Invite new users"),
            ("settings.roles.manage", "Manage roles (create/edit/delete)"),
        ]),
    ],
    inheritance={"manage": ["view"]},   # display hint only (data-implies +
)                                       # a note); enforcement stays app-side
```

### 2) Implement the provider

`AuthzProvider` is the app-side seam — the package never touches a DB.
Implement it over the app's own tables (Books: `organization_users` +
`roles` + `role_permissions`):

```python
from mcp_core.authz_ui import AuthzError, Member, Role

class AppAuthzProvider:
    def can_manage(self, actor):            # the single page gate; fail closed
        ou = current_org_user()
        return ou is not None and (
            ou.role in ("owner", "admin")
            or has_permission(ou, "settings.users.edit")
        )

    def list_members(self, org): ...        # -> [Member(...), ...]
    def get_member(self, org, member_id): ...
    def invite_member(self, org, email, role_id): ...   # app owns email/token flow
    def set_member_role(self, org, member_id, role_id): ...
    def deactivate_member(self, org, member_id): ...
    def reactivate_member(self, org, member_id): ...
    def list_roles(self, org): ...          # -> [Role(...), ...]
    def get_role(self, org, role_id): ...   # None -> 404
    def create_role(self, org, name, description=""): ...
    def set_role_permissions(self, org, role_id, keys): ...  # FULL-set replace
    def delete_role(self, org, role_id): ...
```

Conventions: `org` is opaque (whatever your `org_loader` returns — id, ORM
row, or `None` for single-tenant). Refuse an operation by raising
`AuthzError("user-facing message")` — HTML pages flash it, JSON endpoints
return it as a 400 `{"error": ...}`. Mark hardcoded roles (Books'
owner/admin) `Role(is_system=True)`: they render as "Full access /
built-in", are excluded from the invite form, and the blueprint refuses to
edit or delete them even on a forged form (keep the provider refusing too
— defense in depth). Every matrix save is validated against the catalog
*before* the provider sees it, so unknown/forged keys can never be
written.

### 3) Mount the blueprint

```python
from mcp_core.authz_ui import authz_admin_blueprint

app.register_blueprint(authz_admin_blueprint(
    catalog,
    AppAuthzProvider(),
    base_template="base.html",         # must define a `content` block
    url_prefix="/admin/authz",         # default
    actor_loader=lambda: current_user,        # optional; default passes None
    org_loader=lambda: session.get("org_id"), # optional; default passes None
))
```

Routes (all gated by `provider.can_manage(actor)` -> 403; JSON routes get
a JSON 403 body):

| Route | What it does |
| --- | --- |
| `GET /members` | Member table, invite form, per-row role assign + activate toggle |
| `POST /members/invite`, `POST /members/<id>/role`, `POST /members/<id>/active` | Member ops (form posts, redirect + flash) |
| `GET /roles`, `POST /roles` | Role list + create form |
| `GET /roles/<id>` | Permission **matrix** generated from the catalog: one fieldset per group, one checkbox per key |
| `POST /roles/<id>/permissions` | Saves the FULL checked set (unchecked = revoked) |
| `POST /roles/<id>/delete` | Delete role |
| `GET /api/catalog`, `GET/POST /api/members...`, `GET/POST/PUT/DELETE /api/roles...` | JSON mirror of every operation for JS-driven pages (Books' settings page fetches JSON today) — same gate, same validation |

Templates extend the app-supplied `base_template` and fill its `content`
block. Markup is plain semantic HTML with Bootstrap-ish class names
(`table`, `btn btn-primary`, `badge`) plus `authz-*` hooks — styled apps
look native, unstyled apps degrade fine. Inheritance hints render as
`data-implies="..."` attributes on checkboxes for apps that want Books'
auto-check-view JS behavior.

> **Adoption:** like the MCP migration, each app swaps its existing Users
> & Permissions pages in its own dedicated slice — nothing imports
> `authz_ui` yet. **PM goes first**: it has NO roles UI at all today, so
> it adopts greenfield (declare catalog, implement provider, mount) with
> nothing to swap. Books is the reference UX this generalizes
> (`app/templates/modules/settings.html`); it can keep its JS-driven page
> on the `/api/*` endpoints and retire its hand-built matrix last.

## Adoption guide (per app)

Each app migrates in its own slice, in four steps:

1. **Implement `TokenValidator`** over the app's `oauth_tokens` table
   (`mcp_core.hash_token` matches the existing `OAuthToken.hash_token`
   sha256 digests byte for byte, so no data migration). Service tokens map
   to `TokenContext(kind="service", org_id=...)`; user tokens to
   `TokenContext(kind="user", principal_user_id=...)`.
2. **Implement `CallerResolver`** (and `PrincipalResolver`) over the app's
   users/memberships tables. Books-style: any miss raises `IdentityError`
   (no guest tier). SR-style external callers: wrap the staff resolver with
   `with_external_tiers(...)` and classify `external_known` /
   `external_new` in the app. Keep resolution stateless across requests;
   per-request memoization (Flask `g`) is fine.
3. **Declare tools** via `@registry.tool(name, permission=..., schema=...)`
   using the app's existing permission keys (the same strings the web UI
   enforces). Port Books' `TOOL_PERMISSIONS` entries 1:1 — tuples become
   `AllOf(...)`, `()` becomes `AllOf()`, per-entity re-gates become
   `dynamic_permission`. Pass the app's `has_permission` semantics as
   `permission_evaluator` where owner-full-access / manage->view
   inheritance matter. Mark structural machine-to-machine callbacks
   `identity_exempt=True` — nothing else.
4. **Mount the blueprint**: build an `McpServer` and register
   `streamable_http_blueprint(server)` at `/mcp`. Existing legacy routes
   (`/mcp/sse`, `/mcp/messages`, REST discovery endpoints) can coexist
   during the migration window and be retired once clients move.

> **Migration status:** SR, Books, CRM, DM, and PM all keep their current
> hand-rolled MCP servers until each is migrated in its own dedicated
> slice. Nothing imports `mcp_core` yet; this repo only has to stay
> byte-compatible with the frozen contracts above so those slices are
> drop-in.

## Development

```
python -m pip install -e .[dev]
python -m pytest
```

The test suite is pure library — fake validators/resolvers, no database,
no app imports.