How to Build a Secure MCP Gateway for Loan Applications with Arcade's MCP Gateway Toolkit

# Install Arcade packages
pip install arcade-ai arcade-tdk arcade-serve

# Or install everything including evaluation framework
pip install 'arcade-ai[all]'

# Set up authentication
arcade login

auth:
  providers:
    - id: financial-oauth
      type: oauth2
      provider_id: custom
      client_id: ${env:CLIENT_ID}
      client_secret: ${env:CLIENT_SECRET}
      authorization_url: "https://api.loanprovider.com/oauth/authorize"
      token_url: "https://api.loanprovider.com/oauth/token"
      scopes:
        - "applications.read"
        - "applications.create"
        - "credit.check"

from arcade_tdk import ToolContext, tool
from arcade_tdk.auth import OAuth2
from typing import Annotated

@tool(
    requires_auth=OAuth2(
        provider_id="financial-oauth",
        scopes=["credit.check", "applications.read"]
    )
)
async def check_applicant_credit(
    context: ToolContext,
    application_id: Annotated[str, "Loan application ID"],
    ssn_last_four: Annotated[str, "Last 4 digits of SSN"],
    consent_verified: Annotated[bool, "Applicant consent verified"]
) -> dict:
    """
    Perform credit check for loan application with OAuth authentication.
    """
    # Arcade handles token management automatically
    if not context.authorization or not context.authorization.token:
        raise ValueError("Authorization required for credit check")

    if not consent_verified:
        return {"error": "Applicant consent required"}

    # Access token is available via context
    headers = {
        "Authorization": f"Bearer {context.authorization.token}",
        "X-Application-ID": application_id
    }

    # Make secure API call to credit bureau
    # Your implementation here

    return {
        "score": 750,
        "risk_tier": "low",
        "checked_at": datetime.now().isoformat()
    }

@tool(
    requires_auth=OAuth2(
        provider_id="financial-oauth",
        scopes=["documents.read", "documents.verify"]
    )
)
async def verify_income_documents(
    context: ToolContext,
    document_ids: Annotated[list[str], "Document IDs to verify"],
    document_type: Annotated[str, "W2, PAYSTUB, or TAX_RETURN"]
) -> dict:
    """
    Verify income documentation for loan processing.
    """
    token = context.authorization.token

    # Implement document verification logic
    verification_results = []
    for doc_id in document_ids:
        # Your verification API call here
        result = {"document_id": doc_id, "verified": True}
        verification_results.append(result)

    return {
        "all_verified": all(r["verified"] for r in verification_results),
        "results": verification_results
    }

from arcadepy import Arcade
import os

class LoanProcessingGateway:
    def __init__(self):
        self.client = Arcade()  # Uses ARCADE_API_KEY env variable

    async def authorize_loan_officer(self, officer_email: str):
        """
        Authorize a loan officer to access loan processing tools.
        """
        # Request authorization for loan processing tool
        auth_response = self.client.tools.authorize(
            tool_name="LoanProcessor.CreateApplication",
            user_id=officer_email
        )

        if auth_response.status != "completed":
            print(f"Authorization required: {auth_response.url}")
            # Wait for OAuth completion
            self.client.auth.wait_for_completion(auth_response)

        return {"authorized": True, "officer": officer_email}

    async def process_application(self, officer_email: str, application_data: dict):
        """
        Process loan application with proper authentication.
        """
        # Execute tool with user context
        response = self.client.tools.execute(
            tool_name="LoanProcessor.ProcessApplication",
            input=application_data,
            user_id=officer_email
        )

        return response.output

# Collect all required scopes for loan processing
client = Arcade()
user_id = "loan_officer@bank.com"

# List all loan processing tools
tools = client.tools.list(toolkit="LoanProcessor")

# Collect required scopes
scopes = set()
for tool in tools:
    if tool.requirements.authorization.oauth2.scopes:
        scopes |= set(tool.requirements.authorization.oauth2.scopes)

# Authenticate once with all scopes
auth_response = client.auth.start(
    user_id=user_id,
    scopes=list(scopes),
    provider="financial-oauth"
)

if auth_response.status != "complete":
    print(f"Authorize here: {auth_response.url}")

client.auth.wait_for_completion(auth_response)

# worker.toml
name = "loan-mcp-gateway"
description = "MCP Gateway for loan application processing"

[deployment]
environment = "production"
min_replicas = 2
max_replicas = 10

[toolkits]
loan_processor = { path = "./loan_tools" }

arcade deploy

# Output:
# ✅ Worker 'loan-mcp-gateway' deployed successfully
# Host: https://your-id.server.arcade.dev

FROM python:3.11-slim

WORKDIR /app

# Install Arcade
RUN pip install arcade-ai arcade-serve

# Copy your tools
COPY loan_tools/ ./loan_tools/

# Set environment variables
ENV ARCADE_API_KEY=${ARCADE_API_KEY}

# Run the MCP server
CMD ["arcade", "serve", "--host", "0.0.0.0", "--port", "8080"]

from langchain_arcade import ArcadeToolManager
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph

# Initialize Arcade tools for loan processing
tool_manager = ArcadeToolManager(api_key=os.environ["ARCADE_API_KEY"])
loan_tools = tool_manager.get_tools(toolkits=["LoanProcessor"])

# Create loan processing agent
model = ChatOpenAI(model="gpt-4")
model_with_tools = model.bind_tools(loan_tools)

# Build stateful workflow for loan applications
workflow = StateGraph()

# Add nodes for different stages
workflow.add_node("credit_check", credit_check_node)
workflow.add_node("document_verify", document_verification_node)
workflow.add_node("decision", loan_decision_node)

# Define edges based on authorization and business logic
workflow.add_edge("credit_check", "document_verify")
workflow.add_edge("document_verify", "decision")

loan_processor = workflow.compile()

from arcade_tdk.exceptions import (
    RetryableToolError,
    ContextRequiredToolError,
    UpstreamRateLimitError
)

@tool
async def process_loan_with_retry(context: ToolContext, application_id: str):
    try:
        # Attempt loan processing
        result = await process_loan(application_id)
        return result

    except UpstreamRateLimitError as e:
        # Handle rate limiting from credit bureaus
        raise RetryableToolError(
            "Credit bureau rate limit reached. Please try again in a few minutes.",
            retry_after=60
        )

    except ValueError as e:
        # Request additional context from the user
        raise ContextRequiredToolError(
            "Additional information needed for loan processing",
            required_context=["income_verification", "employment_history"]
        )

@tool
async def loan_decision_with_audit(
    context: ToolContext,
    application_id: str,
    decision: str,
    factors: list[str]
):
    """
    Record loan decision with full audit trail for compliance.
    """
    # Log decision for regulatory compliance
    audit_entry = {
        "application_id": application_id,
        "decision": decision,
        "factors": factors,
        "decided_by": context.authorization.user_info.get("sub"),
        "timestamp": datetime.now().isoformat()
    }

    # Store audit log (implement your storage solution)
    await store_audit_log(audit_entry)

    # Return decision
    return {
        "decision": decision,
        "audit_logged": True,
        "reference_id": audit_entry["timestamp"]
    }

security:
  authentication:
    isolation_mode: strict
    token_exposure: never
    audit_logging: enabled

  rate_limiting:
    enabled: true
    max_requests_per_minute: 100
    per_user_limits: true

  compliance:
    pii_masking: enabled
    data_retention_days: 2555  # 7 years for financial records

@tool
async def health_check(context: ToolContext):
    """
    Health check endpoint for loan processing gateway.
    """
    checks = {
        "oauth_provider": await check_oauth_connection(),
        "credit_bureau_api": await check_credit_api(),
        "database": await check_database(),
        "mcp_server": True
    }

    return {
        "status": "healthy" if all(checks.values()) else "degraded",
        "checks": checks,
        "timestamp": datetime.now().isoformat()
    }

import pytest
from arcade_tdk.test import ToolTestCase

class TestLoanTools(ToolTestCase):
    async def test_credit_check_requires_auth(self):
        """Test that credit check requires authentication."""
        with pytest.raises(ValueError, match="Authorization required"):
            await check_applicant_credit(
                context=self.mock_context(authorized=False),
                application_id="TEST001",
                ssn_last_four="1234",
                consent_verified=True
            )

    async def test_document_verification(self):
        """Test document verification with valid auth."""
        result = await verify_income_documents(
            context=self.mock_context(authorized=True),
            document_ids=["DOC001", "DOC002"],
            document_type="W2"
        )
        assert result["all_verified"] == True