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:

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.

Related MCP server: FastAPI MCP Production Kit

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

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:

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:

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):

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

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):

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.