#
#
Version: 1.0 Date: January 2026 Status: Draft / RFC
This document outlines a strategy for enabling Model Context Protocol (MCP) support in the Modfin API. MCP is an emerging standard that allows AI assistants (ChatGPT, Claude, Cursor, etc.) to interact with external tools and services in a structured, secure manner.
The Modfin codebase already contains a sophisticated function registry (llm/functions.py) with 30+ business operations exposed as typed, documented functions. This white paper proposes leveraging this existing infrastructure to create an MCP-enabled backend with federated identity support, allowing users to authenticate via their preferred AI assistant.
Key Proposals:
llm/functions.registry to MCP toolsThe Model Context Protocol (MCP) is a standardized protocol for AI assistants to interact with external tools and services. It provides:
Modfin’s core value proposition is enabling small businesses to manage their finances through conversational AI. Currently, this is delivered through:
llm/functions.py module exposing business operationsMCP adoption would:
┌─────────────────────────────────────────────────────────────────────┐
│ MCP Clients │
├─────────────┬─────────────┬─────────────┬─────────────┬────────────┤
│ ChatGPT │ Claude │ Cursor │ Copilot │ Custom │
│ (OpenAI) │ (Anthropic) │ │ (Microsoft) │ Clients │
└──────┬──────┴──────┬──────┴──────┬──────┴──────┬──────┴─────┬──────┘
│ │ │ │ │
└─────────────┴─────────────┴─────────────┴────────────┘
│
│ MCP Protocol
│ (JSON-RPC over SSE/WebSocket)
▼
┌─────────────────────────────────────────────────────────────────────┐
│ MCP Servers │
├─────────────┬─────────────┬─────────────┬─────────────┬────────────┤
│ Modfin │ Stripe │ GitHub │ Notion │ ... │
│ (proposed) │ │ │ │ │
└─────────────┴─────────────┴─────────────┴─────────────┴────────────┘
The llm/functions.py module provides a mature foundation for MCP adoption:
# Current decorator pattern
@function(contacts=False, accounts=False)
@utils.composes.ctx(dbs=lambda dbs: dbs.no_expire())
async def create_contact(dbs, chat,
data: contacts.schemas.Data,
tags: list[contacts.consts.Tag],
account_id: Account = None,
):
"""
How to create a contact. At a minimum, provide a name...
"""
Strengths:
registry dictdbs, chat) excluded from schemas@permitted, @subscribed handle authorizationCurrent Function Categories:
| Category | Functions | Description |
|---|---|---|
| User | get_me, update_my_profile |
User profile management |
| Org | get_org, update_org_setting_* |
Organization settings |
| Contacts | get_contacts, create_contact, update_contact, move_contact |
Customer management |
| Services | get_offered_services, create_offered_service, update_offered_service |
Service catalog |
| Connectors | get_connectors, upsert_connector, trigger_connector |
Payment integrations |
| Ledger | get_ledger, get_account_aging |
Financial queries |
| Events | create_invoice, create_payment, create_reversal |
Financial transactions |
| Utility | get_today, get_current_time |
Helper functions |
The auth/ module provides OAuth 2.0 capabilities:
Endpoints:
POST /v1/auth/login - Password grant (legacy)POST /v1/auth/register/email - Email registrationPOST /v1/auth/authorize - Authorization code creationPOST /v1/auth/token - Token exchangePOST /v1/auth/refresh - Token refreshToken Management:
AuthSession modelctx) field for different token typesGaps for MCP:
.well-known/oauth-authorization-server)Functions receive a chat object with session state:
chat = SimpleNamespace(
owner_id="user-123", # Current user ID
org_id="org-456", # Current organization
book_id="book-789", # Ledger book (optional)
state={ # Session state
"timezone": "America/Los_Angeles",
},
)
This object must be constructed from MCP authentication context.
┌─────────────────────────────────────────────────────────────────────┐
│ MCP Client (ChatGPT, Claude, etc.) │
└─────────────────────────┬───────────────────────────────────────────┘
│
│ MCP Protocol (SSE/WebSocket)
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ FastMCP Server Layer │
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ Transport Layer │ │
│ │ • SSE endpoint for streaming │ │
│ │ • WebSocket support for bidirectional │ │
│ │ • HTTP fallback for simple requests │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ Authentication Layer │ │
│ │ • OAuth 2.0 Discovery │ │
│ │ • Token Verification (Modfin tokens) │ │
│ │ • Federated Identity (OpenAI, Anthropic, etc.) │ │
│ │ • Session Management │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ Context Injection Layer │ │
│ │ • Database session management │ │
│ │ • Chat context construction │ │
│ │ • Organization resolution │ │
│ │ • Permission/subscription validation │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ Tool Registry Bridge │ │
│ │ • Maps llm/functions.registry → MCP tools │ │
│ │ • Schema transformation │ │
│ │ • Result formatting │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ Existing Modfin Backend │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ accounts │ │ contacts │ │ ledgers │ │ events │ │ orgs │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ PostgreSQL Database │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘
FastMCP is the recommended framework for Python MCP servers. It provides:
Integration approach: Rather than rewriting functions, we bridge the existing registry:
from fastmcp import FastMCP
from llm import functions
mcp = FastMCP(name="Modfin")
# Bridge existing registry to MCP
for name, fn in functions.registry.items():
mcp.register_tool(
name=name,
func=create_wrapper(fn),
schema=fn.schema,
description=fn.__doc__,
)
Three authentication modes are proposed:
| Mode | Use Case | Flow |
|---|---|---|
| Token Passthrough | Existing Modfin users | MCP client stores Modfin tokens, passes on each request |
| Federated Identity | New users via AI assistants | Verify upstream identity (ChatGPT/Claude), create linked account |
| OAuth Flow | Web-based authorization | Full OAuth 2.0 authorization code flow with consent screen |
The wrapper layer constructs the chat context from MCP authentication:
async def create_wrapper(fn):
async def wrapper(ctx: MCPContext, **kwargs):
chat = SimpleNamespace(
owner_id=ctx.user_id,
org_id=ctx.claims.get('org_id'),
book_id=ctx.claims.get('book_id'),
state=ctx.session_state or {},
)
async with db.session_ctx() as dbs:
return await fn(dbs=dbs, chat=chat, **kwargs)
return wrapper
We propose a new UserIdentity model to support multiple identity providers per user:
┌─────────────────────────────────────────────────────────────────────┐
│ UserIdentity Table │
├──────────┬─────────────┬─────────────────────┬─────────────────────┤
│ user_id │ provider │ external_id │ verified_at │
├──────────┼─────────────┼─────────────────────┼─────────────────────┤
│ usr_001 │ email │ alice@example.com │ 2025-12-01 10:00:00 │
│ usr_001 │ openai │ user-abc123xyz │ 2026-01-06 14:30:00 │
│ usr_001 │ anthropic │ user_019283746 │ 2026-01-06 15:45:00 │
│ usr_002 │ openai │ user-def456uvw │ 2026-01-05 09:00:00 │
│ usr_003 │ email │ bob@example.com │ 2025-11-15 08:00:00 │
└──────────┴─────────────┴─────────────────────┴─────────────────────┘
Benefits:
SQLAlchemy Model Definition:
# auth/consts.py
import enum
import utils
class IdentityProvider(enum.StrEnum):
"""
All supported identity providers
"""
email = 'email'
openai = 'openai'
anthropic = 'anthropic'
google = 'google'
github = 'github'
class IdentityExternalConfig(utils.NamespaceEnum):
"""
Configuration overrides for providers that deviate from defaults.
Only list providers that need special behavior.
Supported attributes:
- case: 'insensitive' for case-insensitive matching (default: sensitive)
"""
email = dict(case='insensitive')
def is_case_insensitive(provider: IdentityProvider | str) -> bool:
"""
Check if provider uses case-insensitive matching.
"""
try:
config = IdentityExternalConfig[provider]
return getattr(config, 'case', None) == 'insensitive'
except KeyError:
return False
# For index generation: list of case-insensitive provider names
CASE_INSENSITIVE_PROVIDERS = [p.name for p in IdentityExternalConfig
if is_case_insensitive(p.name)]
# auth/models.py
import db
import dispatch
from . import consts
@db.register.mapped
@dispatch.model.insert('auth.identity.inserted')
@dispatch.model.update('auth.identity.updated')
class UserIdentity(db.IdMixin, db.TimeMixin, db.CRUD):
"""
Links external identities to internal users.
A user can have multiple identities (email + ChatGPT + Claude).
"""
__tablename__ = 'user_identities'
__table_args__ = (
# Case-insensitive uniqueness for configured providers (e.g., email)
db.Index(
'uq_identity_case_insensitive',
'provider',
db.func.lower(external_id),
unique=True,
postgresql_where=(provider.in_(consts.CASE_INSENSITIVE_PROVIDERS)),
),
# Case-sensitive uniqueness for all other providers
db.Index(
'uq_identity_case_sensitive',
'provider',
'external_id',
unique=True,
postgresql_where=(~provider.in_(consts.CASE_INSENSITIVE_PROVIDERS)),
),
{'schema': 'identity'},
)
user_id: db.Mapped[str] = db.column(db.ForeignKey('identity.users.id'), index=True)
provider: db.Mapped[consts.IdentityProvider] = db.column(db.ENUM(consts.IdentityProvider))
external_id: db.Mapped[str] = db.column() # Case preserved, uniqueness via partial indexes
verified_at: db.Mapped[datetime.datetime | None] = db.column()
meta: db.Mapped[dict | None] = db.column(db.JSONB) # Provider-specific metadata
user: db.Mapped[User] = db.relationship(back_populates='identities')
@classmethod
async def get_by_external(cls, dbs, provider: consts.IdentityProvider, external_id: str):
"""
Lookup identity by provider and external_id.
Case sensitivity determined by provider config.
"""
if consts.is_case_insensitive(provider):
qry = db.select(cls).where(
cls.provider == provider,
db.func.lower(cls.external_id) == external_id.lower(),
)
else:
qry = db.select(cls).where(
cls.provider == provider,
cls.external_id == external_id,
)
r = await dbs.execute(qry)
return r.scalar()
Add the relationship to the existing User model:
# users/models.py (add to User class)
class User(db.IdMixin, db.TimeMixin, db.CRUD):
# ... existing fields ...
identities: db.Mapped[list[UserIdentity]] = db.relationship(back_populates='user')
| Provider | Identifier | Verification Method | Status |
|---|---|---|---|
email |
Email address | Password hash / magic link | Existing |
openai |
ChatGPT user ID | Signed JWT (JWKS verification) | Proposed |
anthropic |
Claude user ID | Signed JWT (JWKS verification) | Proposed |
google |
Google account ID | OAuth 2.0 / OIDC | Future |
github |
GitHub user ID | OAuth 2.0 | Future |
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ MCP Client │ │ Modfin │ │ Database │
└──────┬───────┘ └──────┬───────┘ └──────┬───────┘
│ │ │
│ 1. Tool call with │ │
│ Authorization: │ │
│ Bearer <token> │ │
│───────────────────────>│ │
│ │ │
│ │ 2. Decode & verify │
│ │ JWT token │
│ │ │
│ │ 3. Lookup user │
│ │───────────────────────>│
│ │ │
│ │ 4. User record │
│ │<───────────────────────│
│ │ │
│ │ 5. Execute tool │
│ │ with user context │
│ │ │
│ 6. Tool result │ │
│<───────────────────────│ │
│ │ │
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ MCP Client │ │ Modfin │ │ OpenAI │
│ (ChatGPT) │ │ │ │ JWKS │
└──────┬───────┘ └──────┬───────┘ └──────┬───────┘
│ │ │
│ 1. authenticate() │ │
│ provider: "openai" │ │
│ identity_token: JWT │ │
│───────────────────────>│ │
│ │ │
│ │ 2. Fetch JWKS │
│ │───────────────────────>│
│ │ │
│ │ 3. Public keys │
│ │<───────────────────────│
│ │ │
│ │ 4. Verify JWT │
│ │ signature │
│ │ │
│ │ 5. Extract claims: │
│ │ sub: user-abc123 │
│ │ email: (optional) │
│ │ │
│ │ 6. Lookup/create user │
│ │ in UserIdentity │
│ │ │
│ │ 7. Generate Modfin │
│ │ access token │
│ │ │
│ 8. {access_token, │ │
│ refresh_token, │ │
│ user_id} │ │
│<───────────────────────│ │
│ │ │
Users connecting via a new identity provider can be linked to existing accounts.
Auto-Linking via Provider Email Claim:
MCP OAuth tokens can include the user’s email if we request the openid email scopes.
When the provider (OpenAI, Anthropic) shares the user’s verified email, we can auto-link:
1. User connects from ChatGPT
2. OAuth token includes: {sub: "user-abc123", email: "alice@example.com"}
3. Modfin checks: does alice@example.com exist in UserIdentity?
- Yes → auto-link OpenAI identity to that user (seamless!)
- No → create new user with both OpenAI + email identities
4. Done - no verification needed (provider already verified email)
Fallback: Verification Code Flow
If the provider token lacks an email claim (user didn’t consent or provider doesn’t share it), we fall back to manual verification:
User (via Claude): "Show my invoices"
Modfin: "I don't recognize this Claude account. Would you like to:
1. Create a new Modfin account
2. Link to an existing account"
User: "Link to existing - my email is alice@example.com"
Modfin: "I've sent a verification email to alice@example.com.
Enter the 6-digit code, or click the link in the email."
User: "The code is 847291"
Modfin: "Your Claude identity is now linked to your Modfin account.
You now have 2 linked identities: email, anthropic"
Verification Email Design:
A single email template with conditional rendering supports multiple flows:
| Context | Includes | Use Case |
|---|---|---|
| Web signup | Link only | User clicks to verify |
| Chat linking | Code only | User enters code in chat |
| Flexible | Both link + code | User chooses method |
The verification token encodes both the email and the pending identity to link:
# Token payload
{
"email": "alice@example.com",
"link_provider": "anthropic", # pending identity
"link_external_id": "user_019283", # provider's user ID
"exp": <10 minutes from now>,
}
Both verification methods complete the same action:
GET /v1/auth/verify/{token} → decode, verify email, link identityPOST /v1/auth/verify/code → lookup pending link, verify, link identityThis allows users to verify via whichever method is most convenient for their context.
For frictionless onboarding, users can register directly through MCP:
Email Registration:
@mcp.tool()
async def register_email(
email: str,
password: str,
name: str | None = None,
) -> AuthResult:
"""
Create a new Modfin account with email and password.
"""
Identity-Based Registration:
@mcp.tool()
async def authenticate(
provider: Literal["openai", "anthropic"],
identity_token: str | None = None,
) -> AuthResult:
"""
Authenticate or register using your AI assistant's identity.
If no account exists, one will be created automatically.
"""
Security Controls:
Deliverables:
mcp/ module structurellm/functions.registry to MCP toolsTesting:
Deliverables:
UserIdentity model and migrationsIdentityVerifier for OpenAI/AnthropicTesting:
Deliverables:
Testing:
mcp/
├── __init__.py
├── server.py # FastMCP server initialization
├── auth.py # Authentication and identity verification
├── bridge.py # llm/functions.registry → MCP bridge
├── context.py # Chat context construction
├── tools/
│ ├── __init__.py
│ ├── auth_tools.py # authenticate, register_email, link_identity
│ └── admin_tools.py # MCP-specific admin tools
├── schemas.py # MCP-specific Pydantic schemas
└── tests/
├── test_auth.py
├── test_bridge.py
└── test_tools.py
# mcp/config.py
from pydantic_settings import BaseSettings
class MCPSettings(BaseSettings):
# Server
mcp_enabled: bool = True
mcp_port: int = 8001
mcp_base_url: str = "https://mcp.modfin.com"
# Authentication
mcp_token_expiry_minutes: int = 60
mcp_refresh_expiry_days: int = 30
# Identity Providers
openai_jwks_uri: str = "https://api.openai.com/.well-known/jwks.json"
anthropic_jwks_uri: str = "https://api.anthropic.com/.well-known/jwks.json"
# Rate Limiting
registration_rate_limit: int = 5 # per hour
auth_rate_limit: int = 20 # per minute
# Feature Flags
federated_identity_enabled: bool = True
headless_registration_enabled: bool = True
class Config:
env_prefix = "MODFIN_"
| Threat | Mitigation |
|---|---|
| Token theft | Short-lived access tokens (1 hour), refresh rotation |
| Identity spoofing | Cryptographic verification via JWKS |
| Brute force registration | Rate limiting, CAPTCHA for suspicious patterns |
| Privilege escalation | Existing @permitted decorators enforced |
| Data exfiltration | Query limits, audit logging |
| Replay attacks | Token nonces, expiration validation |
Federated identity tokens MUST be verified cryptographically:
async def verify_identity_token(provider: str, token: str) -> Claims:
# 1. Fetch provider's JWKS (cached)
jwks = await fetch_jwks(JWKS_URIS[provider])
# 2. Decode header to get key ID
header = jwt.get_unverified_header(token)
# 3. Find matching key
key = find_key(jwks, header['kid'])
if not key:
raise SecurityError("Unknown signing key")
# 4. Verify signature, issuer, audience, expiration
claims = jwt.decode(
token,
key,
algorithms=['RS256', 'ES256'],
issuer=ISSUERS[provider],
audience=MODFIN_CLIENT_ID,
options={'require': ['exp', 'iat', 'sub']},
)
return claims
Rate limiting uses the @decorators.ratelimit decorator, which checks eagerly when
the function is called. It leverages utils.limit (with raises parameter) and
utils.reiterable for efficient iterator handling.
# auth/models.py
@classmethod
@decorators.ratelimit(
window=consts.RESET_PASSWORD_THROTTLE_SECS, # 5 minutes
when=lambda obj: obj.created_at,
message='Recently sent reset password link',
)
async def validate_ratelimit(cls, dbs, lookup):
"""
Validate rate limit for lookup, raises if recently used.
"""
return await cls.get_by_lookup(dbs, lookup)
| Parameter | Description |
|---|---|
window |
Time window in seconds |
when |
Lambda to extract timestamp from object(s) |
limit |
Max count within window (None = any within window) |
scalar |
True for single object, False for iterable (default) |
exc |
Exception to raise (default: HTTPException 429) |
message |
Error message when rate limited |
| Action Type | Window | Limit | Notes |
|---|---|---|---|
| Password reset | 5 min | 1 | Throttle per email |
| Authentication | 1 min | 20 | Per identity |
| Registration | 1 hour | 5 | Per email domain |
| Read operations | 1 min | 100 | Per user |
| Write operations | 1 min | 30 | Per user |
| Financial ops | 1 hour | 60 | Per org |
All MCP operations should be logged for security audit:
@dataclass
class MCPAuditLog:
timestamp: datetime
user_id: str | None
identity_provider: str | None
tool_name: str
tool_args: dict # Sanitized, no PII
result_status: str # success, error, denied
client_info: dict # MCP client metadata
ip_address: str
New Table: user_identities
The table is auto-created from the SQLAlchemy model defined in Section 4.1. For
reference, the equivalent raw SQL (generated via ./tool.py db schema diff):
CREATE TABLE identity.user_identities (
id VARCHAR(12) PRIMARY KEY DEFAULT encode(gen_random_bytes(6), 'hex'),
user_id VARCHAR NOT NULL REFERENCES identity.users(id),
provider identityprovider NOT NULL,
external_id VARCHAR NOT NULL,
verified_at TIMESTAMP WITH TIME ZONE,
meta JSONB,
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL,
updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL
);
CREATE INDEX ix_identity_user_identities_user_id ON identity.user_identities(user_id);
-- Partial unique indexes for provider-aware case sensitivity
-- Case-insensitive providers (currently just 'email')
CREATE UNIQUE INDEX uq_identity_case_insensitive
ON identity.user_identities (provider, LOWER(external_id))
WHERE provider = 'email';
-- Case-sensitive providers (all others)
CREATE UNIQUE INDEX uq_identity_case_sensitive
ON identity.user_identities (provider, external_id)
WHERE provider != 'email';
Case Sensitivity Design: Rather than using CITEXT globally, we use partial unique indexes to handle case sensitivity per provider:
LOWER(external_id). This matches the existing
user_emails.email behavior where Alice@Example.com equals alice@example.com.user-ABC123
and user-abc123 as distinct users).The UserIdentity.get_by_external() class method (defined in Section 4.1) handles
the case normalization in application code.
Backfill Existing Users:
One-time migration to create email identities for existing users:
-- Backfill email identities from verified user_emails
-- Run once after table creation
INSERT INTO identity.user_identities (user_id, provider, external_id, verified_at)
SELECT DISTINCT ON (ue.owner_id)
ue.owner_id,
'email',
ue.email, -- Preserve original case; uniqueness handled by partial index
ue.updated_at
FROM identity.user_emails ue
WHERE ue.status = 'verified'
ORDER BY ue.owner_id, ue.updated_at DESC
ON CONFLICT DO NOTHING;
The MCP server operates as an additional interface; existing systems continue unchanged:
Gradual rollout via feature flags:
# Controlled rollout
if settings.mcp_enabled:
app.mount("/mcp", mcp.get_app())
if settings.federated_identity_enabled:
mcp.register_tool(authenticate)
mcp.register_tool(link_identity)
if settings.headless_registration_enabled:
mcp.register_tool(register_email)
org_id parameter)set_org tooldelete_org)Pricing: How to meter MCP usage?
MCP uses the existing subscription system - no separate billing needed:
Subscription.owner_id references the orgUser.metadata_.stripe.customer_idllm_chat is already a feature - defined in subscriptions.consts.Feature@subscribed decorator - already validates features, usage limits, and seatsFor MCP, the flow is:
subscriptions.stripe.get_or_create_customer)@subscribed(feature=Feature.llm_chat) validation# Existing decorator in llm/functions.py already handles this
@function(events=True)
@subscribed(usage=True) # Validates subscription limits
async def create_invoice(dbs, chat, ...):
The “correct Stripe customer token” is simply ensuring User.metadata_.stripe.customer_id
is set, which happens automatically via the existing Stripe integration when users
subscribe through any channel.
powered_by: "Modfin" in error responses| Term | Definition |
|---|---|
| MCP | Model Context Protocol - standard for AI tool integration |
| FastMCP | Python framework for building MCP servers |
| JWKS | JSON Web Key Set - public keys for JWT verification |
| Federated Identity | Using external identity providers for authentication |
| Tool | MCP term for a callable function/capability |
| Context | MCP session state and metadata |
This document is a living specification. Please submit feedback and revisions via pull request.