Skip to content

MCP‐Implementation‐Plans

Jump to bottom
Jeremy Nelson edited this page Feb 2, 2026 · 1 revision

FOLIO MCP Server - Detailed Implementation Plans

This document provides comprehensive implementation plans for three architectural approaches to exposing FOLIO API endpoints (1,749 endpoints) through the Model Context Protocol (MCP) server.

Project Context:

  • Existing: FastAPI app with fastapi-mcp@0.4.0 integration
  • Current branch: mcp-apis
  • Base URL: OKAPI_URL environment variable
  • Authentication: Via folioclient library using TENANT_ID/ADMIN_USER/ADMIN_PASSWORD

Option 1: Dynamic Proxy/Passthrough Pattern

Overview

Lightweight FastAPI router that dynamically proxies all FOLIO API requests through Okapi gateway, automatically exposing them as MCP tools.

Architecture Diagram

┌─────────────┐
│  AI Agent   │
│  (Claude)   │
└──────┬──────┘
       │ MCP Protocol
       ▼
┌─────────────────────────────────────────┐
│  FastAPI-MCP Server (edge-ai)           │
│  ┌───────────────────────────────────┐  │
│  │  /folio-api/{module}/{endpoint}   │  │
│  │  ├─ Dynamic Route Handler         │  │
│  │  │  • Extract path/method/body    │  │
│  │  │  • Add auth headers            │  │
│  │  │  • Forward to Okapi            │  │
│  │  └─ Return proxied response       │  │
│  └───────────────────────────────────┘  │
└──────────────┬──────────────────────────┘
               │ HTTP
               ▼
        ┌──────────────┐
        │ FOLIO Okapi  │
        │   Gateway    │
        └──────────────┘

File Structure

src/edge_ai/
├── main.py                              # Update to include new router
├── folio_proxy/
│   ├── __init__.py                      # Module initialization
│   ├── router.py                        # Main proxy router
│   ├── models.py                        # Generic request/response models
│   ├── middleware.py                    # Auth & header management
│   └── utils.py                         # Helper functions
└── tests/
    └── folio_proxy/
        ├── test_proxy_router.py         # Integration tests
        └── test_middleware.py           # Middleware tests

Detailed Implementation

Step 1: Create Proxy Router (src/edge_ai/folio_proxy/router.py)

"""
FOLIO API Proxy Router
Dynamically proxies all FOLIO API endpoints through Okapi gateway.
"""
from fastapi import APIRouter, Request, Response, HTTPException, Header
from typing import Optional, Dict, Any
import httpx
import os
from enum import Enum

router = APIRouter(prefix="/folio-api", tags=["folio-proxy"])

# FOLIO module categories for better MCP tool organization
class FOLIOModule(str, Enum):
    INVENTORY = "inventory"
    CIRCULATION = "circulation"
    ACQUISITIONS = "acquisitions"
    FINANCE = "finance"
    USERS = "users"
    ORGANIZATIONS = "organizations"
    ORDERS = "orders"
    INVOICES = "invoices"
    FEES_FINES = "fees-fines"
    CALENDAR = "calendar"
    CONFIGURATION = "configuration"
    AUDIT = "audit"


@router.api_route(
    "/{module}/{path:path}",
    methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
    operation_id="folio_proxy_{module}_{path}",
    summary="Proxy FOLIO API request",
    description="""
    Dynamically proxies FOLIO API requests to the Okapi gateway.

    **Authentication**: Automatically handled via FOLIO credentials.

    **Modules**: Use FOLIO module names (e.g., 'inventory-storage', 'circulation', etc.)

    **Path**: Full endpoint path after module (e.g., 'instances', 'loans/123')

    **Examples**:
    - GET /folio-api/inventory-storage/instances
    - POST /folio-api/circulation/check-out-by-barcode
    - PUT /folio-api/orders/composite-orders/abc-123
    """,
)
async def folio_proxy(
    module: str,
    path: str,
    request: Request,
    x_okapi_tenant: Optional[str] = Header(None),
    x_okapi_token: Optional[str] = Header(None),
) -> Dict[str, Any]:
    """
    Universal FOLIO API proxy handler.

    Args:
        module: FOLIO module name (e.g., 'inventory-storage')
        path: Endpoint path within module
        request: FastAPI request object
        x_okapi_tenant: Optional Okapi tenant override
        x_okapi_token: Optional Okapi token override

    Returns:
        JSON response from FOLIO API

    Raises:
        HTTPException: If FOLIO API request fails
    """
    # Get FOLIO connection details
    okapi_url = os.getenv("OKAPI_URL")
    tenant_id = x_okapi_tenant or os.getenv("TENANT_ID")

    if not okapi_url or not tenant_id:
        raise HTTPException(
            status_code=500,
            detail="FOLIO configuration missing (OKAPI_URL or TENANT_ID)"
        )

    # Build full FOLIO endpoint URL
    folio_endpoint = f"/{module}/{path}"
    full_url = f"{okapi_url.rstrip('/')}{folio_endpoint}"

    # Get request body if present
    body = None
    if request.method in ["POST", "PUT", "PATCH"]:
        body = await request.json() if await request.body() else None

    # Get query parameters
    query_params = dict(request.query_params)

    # Prepare headers
    headers = await _prepare_folio_headers(tenant_id, x_okapi_token)

    # Make request to FOLIO
    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            response = await client.request(
                method=request.method,
                url=full_url,
                json=body,
                params=query_params,
                headers=headers,
            )

            # Return response
            if response.status_code >= 400:
                raise HTTPException(
                    status_code=response.status_code,
                    detail={
                        "message": "FOLIO API request failed",
                        "folio_endpoint": folio_endpoint,
                        "folio_response": response.text,
                    }
                )

            return response.json() if response.content else {}

        except httpx.RequestError as e:
            raise HTTPException(
                status_code=503,
                detail=f"Failed to connect to FOLIO: {str(e)}"
            )


async def _prepare_folio_headers(
    tenant_id: str,
    token: Optional[str] = None
) -> Dict[str, str]:
    """
    Prepare headers for FOLIO API request.

    Args:
        tenant_id: FOLIO tenant ID
        token: Optional pre-existing Okapi token

    Returns:
        Dictionary of headers for FOLIO request
    """
    headers = {
        "X-Okapi-Tenant": tenant_id,
        "Content-Type": "application/json",
        "Accept": "application/json",
    }

    # Get or generate Okapi token
    if token:
        headers["X-Okapi-Token"] = token
    else:
        # Use folioclient to get token
        from folioclient import FolioClient

        client = FolioClient(
            os.getenv("OKAPI_URL"),
            tenant_id,
            os.getenv("ADMIN_USER"),
            os.getenv("ADMIN_PASSWORD"),
        )

        # FolioClient handles authentication internally
        # Access the token from the client's session
        if hasattr(client, 'okapi_token'):
            headers["X-Okapi-Token"] = client.okapi_token

    return headers


# Add convenience endpoints for common operations
@router.get(
    "/search/{module}",
    operation_id="folio_search",
    summary="Search FOLIO resources",
    description="Search any FOLIO module with CQL query",
)
async def folio_search(
    module: str,
    query: str,
    limit: int = 10,
    offset: int = 0,
    request: Request = None,
) -> Dict[str, Any]:
    """
    Convenience endpoint for CQL searches.

    Args:
        module: FOLIO module to search (e.g., 'inventory-storage/instances')
        query: CQL query string
        limit: Maximum results
        offset: Pagination offset

    Returns:
        Search results from FOLIO
    """
    # Construct search request
    search_path = f"{module}?query={query}&limit={limit}&offset={offset}"

    # Use the proxy endpoint
    return await folio_proxy(
        module=module.split('/')[0],
        path='/'.join(module.split('/')[1:]) + f"?query={query}&limit={limit}&offset={offset}",
        request=request,
    )

Step 2: Create Models (src/edge_ai/folio_proxy/models.py)

"""
Generic models for FOLIO proxy responses.
"""
from pydantic import BaseModel, Field
from typing import Any, Dict, List, Optional
from datetime import datetime


class FOLIOProxyRequest(BaseModel):
    """Generic FOLIO API request."""

    module: str = Field(
        description="FOLIO module name (e.g., 'inventory-storage')"
    )
    endpoint: str = Field(
        description="API endpoint path within module"
    )
    method: str = Field(
        default="GET",
        description="HTTP method"
    )
    body: Optional[Dict[str, Any]] = Field(
        default=None,
        description="Request body for POST/PUT/PATCH"
    )
    query_params: Optional[Dict[str, str]] = Field(
        default=None,
        description="Query parameters"
    )


class FOLIOProxyResponse(BaseModel):
    """Generic FOLIO API response."""

    success: bool = Field(
        description="Whether the request succeeded"
    )
    status_code: int = Field(
        description="HTTP status code"
    )
    data: Optional[Dict[str, Any]] = Field(
        default=None,
        description="Response data from FOLIO"
    )
    error: Optional[str] = Field(
        default=None,
        description="Error message if request failed"
    )
    folio_endpoint: str = Field(
        description="Full FOLIO endpoint that was called"
    )
    timestamp: datetime = Field(
        default_factory=datetime.utcnow,
        description="Timestamp of the request"
    )


class FOLIOSearchRequest(BaseModel):
    """CQL search request."""

    module: str = Field(
        description="Module to search (e.g., 'inventory-storage/instances')"
    )
    query: str = Field(
        description="CQL query string",
        examples=["title=*Shakespeare*", "barcode==12345"]
    )
    limit: int = Field(
        default=10,
        ge=1,
        le=1000,
        description="Maximum results"
    )
    offset: int = Field(
        default=0,
        ge=0,
        description="Pagination offset"
    )


class FOLIOSearchResponse(BaseModel):
    """Search results from FOLIO."""

    total_records: int = Field(
        description="Total matching records"
    )
    records: List[Dict[str, Any]] = Field(
        description="Result records"
    )
    result_info: Optional[Dict[str, Any]] = Field(
        default=None,
        description="Additional result metadata"
    )

Step 3: Update Main App (src/edge_ai/main.py)

# Add to existing imports
from edge_ai.folio_proxy.router import router as folio_proxy_router

# Add after existing routers
app.include_router(folio_proxy_router)

# Update MCP configuration to control tool exposure
from fastapi_mcp import FastApiMCP

# Option A: Expose all proxy endpoints as MCP tools
mcp = FastApiMCP(app)
mcp.mount_http()  # Use HTTP transport (recommended)

# Option B: Selective exposure (recommended to reduce tool count)
mcp = FastApiMCP(
    app,
    include_tags=["folio-proxy"],  # Only expose FOLIO proxy endpoints
    # OR use exclude_tags to hide certain endpoints
    # exclude_tags=["internal", "admin"],
)
mcp.mount_http()

Step 4: Add Configuration (src/edge_ai/folio_proxy/__init__.py)

"""
FOLIO API Proxy Module

Provides dynamic proxying of FOLIO API endpoints through Okapi gateway.
All endpoints are automatically exposed as MCP tools via FastAPI-MCP.
"""

__version__ = "0.1.0"

from .router import router

__all__ = ["router"]

Step 5: Create Tests (tests/folio_proxy/test_proxy_router.py)

"""
Tests for FOLIO proxy router.
"""
import pytest
from fastapi.testclient import TestClient
from unittest.mock import patch, AsyncMock, MagicMock
import httpx


@pytest.fixture
def mock_folio_response():
    """Mock successful FOLIO API response."""
    return {
        "instances": [
            {
                "id": "123-456",
                "title": "Test Book",
                "contributors": [{"name": "Test Author"}]
            }
        ],
        "totalRecords": 1
    }


@pytest.mark.asyncio
async def test_proxy_get_request(client: TestClient, mock_folio_response, monkeypatch):
    """Test proxying a GET request to FOLIO."""

    # Mock environment variables
    monkeypatch.setenv("OKAPI_URL", "https://test-okapi.folio.org")
    monkeypatch.setenv("TENANT_ID", "test-tenant")
    monkeypatch.setenv("ADMIN_USER", "admin")
    monkeypatch.setenv("ADMIN_PASSWORD", "password")

    # Mock httpx.AsyncClient
    mock_response = MagicMock()
    mock_response.status_code = 200
    mock_response.json.return_value = mock_folio_response
    mock_response.content = b'{"instances": []}'

    with patch("httpx.AsyncClient") as mock_client:
        mock_client.return_value.__aenter__.return_value.request = AsyncMock(
            return_value=mock_response
        )

        # Make request
        response = client.get("/folio-api/inventory-storage/instances?limit=10")

        assert response.status_code == 200
        assert response.json() == mock_folio_response


@pytest.mark.asyncio
async def test_proxy_post_request(client: TestClient, monkeypatch):
    """Test proxying a POST request to FOLIO."""

    monkeypatch.setenv("OKAPI_URL", "https://test-okapi.folio.org")
    monkeypatch.setenv("TENANT_ID", "test-tenant")

    mock_response = MagicMock()
    mock_response.status_code = 201
    mock_response.json.return_value = {"id": "new-instance-123"}
    mock_response.content = b'{"id": "new-instance-123"}'

    with patch("httpx.AsyncClient") as mock_client:
        mock_client.return_value.__aenter__.return_value.request = AsyncMock(
            return_value=mock_response
        )

        payload = {
            "title": "New Book",
            "instanceTypeId": "123"
        }

        response = client.post(
            "/folio-api/inventory-storage/instances",
            json=payload
        )

        assert response.status_code == 201


@pytest.mark.asyncio
async def test_proxy_error_handling(client: TestClient, monkeypatch):
    """Test error handling for failed FOLIO requests."""

    monkeypatch.setenv("OKAPI_URL", "https://test-okapi.folio.org")
    monkeypatch.setenv("TENANT_ID", "test-tenant")

    mock_response = MagicMock()
    mock_response.status_code = 404
    mock_response.text = "Not found"

    with patch("httpx.AsyncClient") as mock_client:
        mock_client.return_value.__aenter__.return_value.request = AsyncMock(
            return_value=mock_response
        )

        response = client.get("/folio-api/inventory-storage/instances/nonexistent")

        assert response.status_code == 404
        assert "FOLIO API request failed" in response.json()["detail"]["message"]


@pytest.mark.asyncio
async def test_search_endpoint(client: TestClient, mock_folio_response, monkeypatch):
    """Test convenience search endpoint."""

    monkeypatch.setenv("OKAPI_URL", "https://test-okapi.folio.org")
    monkeypatch.setenv("TENANT_ID", "test-tenant")

    mock_response = MagicMock()
    mock_response.status_code = 200
    mock_response.json.return_value = mock_folio_response
    mock_response.content = b'{"instances": []}'

    with patch("httpx.AsyncClient") as mock_client:
        mock_client.return_value.__aenter__.return_value.request = AsyncMock(
            return_value=mock_response
        )

        response = client.get(
            "/folio-api/search/inventory-storage/instances",
            params={"query": "title=*Test*", "limit": 5}
        )

        assert response.status_code == 200

Implementation Steps

Phase 1: Core Proxy (2-3 days)

  1. Day 1: Setup

    • Create src/edge_ai/folio_proxy/ directory
    • Implement router.py with basic proxy handler
    • Implement models.py with generic request/response models
    • Update main.py to include new router
    • Test with 3-5 sample FOLIO endpoints

  2. Day 2: Enhancement

    • Add comprehensive error handling
    • Implement header management and token refresh
    • Add search convenience endpoint
    • Create integration tests
    • Document API in OpenAPI schema

  3. Day 3: MCP Integration

    • Configure FastAPI-MCP to expose proxy endpoints
    • Test MCP tool discovery
    • Create usage examples
    • Update README with proxy documentation

Phase 2: Optimization (1-2 days)

  1. Optional Enhancements
    • Add response caching for GET requests
    • Implement request rate limiting
    • Add request/response logging
    • Create endpoint catalog/discovery mechanism
    • Add batch request support

Testing Strategy

Unit Tests

# Test proxy router
pytest tests/folio_proxy/test_proxy_router.py -v

# Test with coverage
pytest tests/folio_proxy/ --cov=src/edge_ai/folio_proxy

Integration Tests

# Test against real FOLIO instance (requires .env)
FOLIO_INTEGRATION_TEST=true pytest tests/folio_proxy/ -v

MCP Tool Testing

# Test MCP tool exposure
from fastapi.testclient import TestClient
from edge_ai.main import app

client = TestClient(app)

# Get MCP tool list
response = client.get("/mcp/tools")
assert "folio_proxy" in [tool["name"] for tool in response.json()["tools"]]

Deployment Considerations

Environment Variables

# Required
OKAPI_URL=https://okapi-bugfest.folio.org
TENANT_ID=diku
ADMIN_USER=diku_admin
ADMIN_PASSWORD=admin

# Optional
FOLIO_REQUEST_TIMEOUT=30  # Request timeout in seconds
FOLIO_CACHE_TTL=300       # Cache TTL for GET requests

Docker Deployment

# No changes needed - existing Dockerfile works
# Proxy router automatically included when main.py imports it

Performance Considerations

  • Concurrent Requests: httpx.AsyncClient handles connection pooling
  • Timeouts: Set timeout=30.0 to prevent hanging requests
  • Rate Limiting: Consider implementing rate limiting for production
  • Caching: Add Redis caching for frequently accessed GET endpoints

Effort Estimate

Task Time Complexity
Core proxy router 4 hours Low
Models & middleware 2 hours Low
Error handling 2 hours Medium
Testing 4 hours Medium
Documentation 2 hours Low
Total 14 hours (~2 days) Low-Medium

Pros & Cons Summary

Pros ✅

  • Speed: 2-day implementation
  • Coverage: All 1,749 endpoints immediately available
  • Simplicity: Minimal code to maintain
  • Automatic updates: FOLIO API changes reflected immediately
  • Low risk: Simple passthrough logic

Cons ❌

  • Tool overload: 1,749 MCP tools may overwhelm AI agents
  • No validation: Limited request/response validation
  • No AI assistance: Just raw API access
  • Limited error handling: Basic HTTP error forwarding
  • No semantic understanding: Tools are low-level CRUD operations

Option 2: Domain-Specific Agent Pattern

Overview

Extend existing Pydantic AI agent pattern to create specialized agents for each FOLIO domain (Circulation, Acquisitions, Finance, etc.), exposing high-level semantic operations as MCP tools.

Architecture Diagram

┌─────────────┐
│  AI Agent   │
│  (Claude)   │
└──────┬──────┘
       │ MCP Protocol
       │ "Check out book to patron John (barcode: 12345)"
       ▼
┌──────────────────────────────────────────────────┐
│  FastAPI-MCP Server (edge-ai)                    │
│  ┌────────────────────────────────────────────┐  │
│  │  Domain Routers (Semantic Operations)     │  │
│  │  ├─ POST /circulation/checkout            │  │
│  │  ├─ POST /acquisitions/create_order       │  │
│  │  ├─ POST /finance/process_invoice         │  │
│  │  └─ POST /inventory/catalog_item          │  │
│  └────────────┬───────────────────────────────┘  │
│               │                                   │
│  ┌────────────▼───────────────────────────────┐  │
│  │  Pydantic AI Agents (Business Logic)      │  │
│  │  ├─ CirculationAgent                      │  │
│  │  │  • validate_patron()                   │  │
│  │  │  • check_item_availability()           │  │
│  │  │  • create_loan()                       │  │
│  │  ├─ AcquisitionsAgent                     │  │
│  │  │  • validate_vendor()                   │  │
│  │  │  • calculate_totals()                  │  │
│  │  │  • create_po()                         │  │
│  │  └─ FinanceAgent                          │  │
│  │     • validate_invoice()                  │  │
│  │     • match_to_po()                       │  │
│  │     • approve_payment()                   │  │
│  └────────────┬───────────────────────────────┘  │
│               │                                   │
│  ┌────────────▼───────────────────────────────┐  │
│  │  Agent Tools (FOLIO API Calls)            │  │
│  │  └─ FolioClient wrapper methods           │  │
│  └────────────┬───────────────────────────────┘  │
└───────────────┼───────────────────────────────────┘
                │ HTTP
                ▼
         ┌──────────────┐
         │ FOLIO Okapi  │
         │   Gateway    │
         └──────────────┘

File Structure

src/edge_ai/
├── main.py                                    # Update to include domain routers
├── circulation/
│   ├── __init__.py
│   ├── router.py                              # Circulation endpoints
│   ├── agents/
│   │   ├── checkout_agent.py                  # Check-out/check-in workflows
│   │   ├── request_agent.py                   # Patron requests
│   │   └── renewal_agent.py                   # Loan renewals
│   ├── models/
│   │   ├── loan.py                            # Loan Pydantic models
│   │   ├── request.py                         # Request models
│   │   └── patron.py                          # Patron models
│   └── tools/
│       └── folio_circulation.py               # FOLIO API tools
├── acquisitions/
│   ├── __init__.py
│   ├── router.py                              # Acquisitions endpoints
│   ├── agents/
│   │   ├── order_agent.py                     # Purchase order workflows
│   │   ├── receiving_agent.py                 # Receiving workflows
│   │   └── vendor_agent.py                    # Vendor management
│   ├── models/
│   │   ├── order.py                           # Order models
│   │   ├── vendor.py                          # Vendor models
│   │   └── receiving.py                       # Receiving models
│   └── tools/
│       └── folio_acquisitions.py              # FOLIO API tools
├── finance/
│   ├── __init__.py
│   ├── router.py                              # Finance endpoints
│   ├── agents/
│   │   ├── invoice_agent.py                   # Invoice processing (EXISTING)
│   │   ├── payment_agent.py                   # Payment workflows
│   │   └── budget_agent.py                    # Budget management
│   ├── models/
│   │   ├── invoice.py                         # Invoice models
│   │   ├── payment.py                         # Payment models
│   │   └── budget.py                          # Budget models
│   └── tools/
│       └── folio_finance.py                   # FOLIO API tools
├── users/
│   ├── __init__.py
│   ├── router.py                              # User management endpoints
│   ├── agents/
│   │   ├── patron_agent.py                    # Patron account management
│   │   └── permission_agent.py                # Permission workflows
│   ├── models/
│   │   ├── user.py                            # User models
│   │   └── permission.py                      # Permission models
│   └── tools/
│       └── folio_users.py                     # FOLIO API tools
├── shared/
│   ├── __init__.py
│   ├── agent_base.py                          # Base agent class
│   ├── folio_client_wrapper.py                # Enhanced FolioClient
│   └── validation.py                          # Common validation logic
└── tests/
    ├── circulation/
    ├── acquisitions/
    ├── finance/
    └── users/

Detailed Implementation

Step 1: Create Base Agent Class (src/edge_ai/shared/agent_base.py)

"""
Base class for domain-specific FOLIO agents.
Provides common functionality for all agents.
"""
from dataclasses import dataclass
from typing import Optional, TypeVar, Generic, Any, Dict
from pydantic_ai import Agent, RunContext
from pydantic import BaseModel
import os
from folioclient import FolioClient


T = TypeVar('T', bound=BaseModel)


@dataclass
class BaseFOLIODependencies:
    """Base dependencies for all FOLIO agents."""

    folio_client: Optional[FolioClient] = None
    tenant_id: Optional[str] = None
    user_context: Optional[Dict[str, Any]] = None

    def __post_init__(self):
        """Initialize FOLIO client if not provided."""
        if self.folio_client is None:
            self.folio_client = FolioClient(
                os.getenv("OKAPI_URL"),
                os.getenv("TENANT_ID"),
                os.getenv("ADMIN_USER"),
                os.getenv("ADMIN_PASSWORD"),
            )

        if self.tenant_id is None:
            self.tenant_id = os.getenv("TENANT_ID")


class BaseFOLIOAgent(Generic[T]):
    """
    Base class for FOLIO domain agents.

    Provides:
    - Common agent setup
    - Standard error handling
    - FOLIO client integration
    - Logging and observability
    """

    def __init__(
        self,
        agent_name: str,
        output_type: type[T],
        system_prompt: str,
        retries: int = 3,
    ):
        self.agent_name = agent_name
        self.output_type = output_type
        self.system_prompt_text = system_prompt

        # Create Pydantic AI agent
        self.agent = Agent(
            name=agent_name,
            deps_type=BaseFOLIODependencies,
            output_type=output_type,
            retries=retries,
        )

        # Register system prompt
        @self.agent.system_prompt
        async def system_prompt(ctx: RunContext[BaseFOLIODependencies]) -> str:
            return self.system_prompt_text

    def register_tool(self, func):
        """Decorator to register agent tools."""
        return self.agent.tool(retries=2)(func)

    async def run(
        self,
        prompt: str,
        deps: Optional[BaseFOLIODependencies] = None,
        **kwargs
    ) -> T:
        """
        Run the agent with given prompt.

        Args:
            prompt: User prompt/instruction
            deps: Agent dependencies
            **kwargs: Additional arguments for agent.run()

        Returns:
            Agent output of type T
        """
        if deps is None:
            deps = BaseFOLIODependencies()

        result = await self.agent.run(prompt, deps=deps, **kwargs)
        return result.output

Step 2: Create Circulation Agent (src/edge_ai/circulation/agents/checkout_agent.py)

"""
Circulation Checkout Agent
Handles patron check-out workflows with AI assistance.
"""
from dataclasses import dataclass
from typing import Optional, List, Dict, Any
from pydantic import BaseModel, Field
from pydantic_ai import RunContext
from datetime import datetime, timedelta
import os

from edge_ai.shared.agent_base import BaseFOLIOAgent, BaseFOLIODependencies
from edge_ai.circulation.models.loan import Loan, CheckoutRequest, CheckoutResponse


@dataclass
class CirculationDependencies(BaseFOLIODependencies):
    """Dependencies specific to circulation operations."""

    service_point_id: Optional[str] = None
    proxy_enabled: bool = True


CHECKOUT_SYSTEM_PROMPT = """
You are an expert circulation librarian assistant for a FOLIO library system.

Your role is to help facilitate patron checkouts by:
1. Validating patron eligibility (active account, no blocks)
2. Checking item availability (not already checked out, not restricted)
3. Applying circulation policies (loan periods, renewals)
4. Creating accurate loan records
5. Providing helpful messages to staff

Always prioritize patron service while following library policies.
Be clear about why checkouts cannot be completed if there are blockers.
"""


class CheckoutAgent(BaseFOLIOAgent[CheckoutResponse]):
    """Agent for handling patron checkout operations."""

    def __init__(self):
        super().__init__(
            agent_name="circulation_checkout",
            output_type=CheckoutResponse,
            system_prompt=CHECKOUT_SYSTEM_PROMPT,
            retries=3,
        )

        # Register circulation-specific tools
        self._register_circulation_tools()

    def _register_circulation_tools(self):
        """Register tools for circulation operations."""

        @self.register_tool
        async def get_patron_by_barcode(
            ctx: RunContext[CirculationDependencies],
            barcode: str
        ) -> Dict[str, Any]:
            """
            Retrieve patron information by barcode.

            Args:
                barcode: Patron barcode

            Returns:
                Patron record with blocks and permissions
            """
            client = ctx.deps.folio_client

            # Search for user by barcode
            query = f'barcode=="{barcode}"'
            users = client.folio_get("/users", query=query)

            if not users.get("users"):
                return {"error": f"No patron found with barcode {barcode}"}

            user = users["users"][0]
            user_id = user["id"]

            # Check for patron blocks
            blocks = client.folio_get(
                f"/automated-patron-blocks/{user_id}",
                key="automatedPatronBlocks"
            )

            # Get manual blocks
            manual_blocks = client.folio_get(
                "/manualblocks",
                query=f'userId=="{user_id}"',
                key="manualblocks"
            )

            return {
                "user": user,
                "automated_blocks": blocks,
                "manual_blocks": manual_blocks,
                "is_blocked": len(blocks) > 0 or len(manual_blocks) > 0
            }

        @self.register_tool
        async def get_item_by_barcode(
            ctx: RunContext[CirculationDependencies],
            barcode: str
        ) -> Dict[str, Any]:
            """
            Retrieve item information by barcode.

            Args:
                barcode: Item barcode

            Returns:
                Item record with availability status
            """
            client = ctx.deps.folio_client

            # Search for item
            query = f'barcode=="{barcode}"'
            items = client.folio_get(
                "/item-storage/items",
                query=query
            )

            if not items.get("items"):
                return {"error": f"No item found with barcode {barcode}"}

            item = items["items"][0]

            # Check if already checked out
            is_available = item.get("status", {}).get("name") == "Available"

            # Get loan policy
            # (In production, would use circulation rules to determine policy)

            return {
                "item": item,
                "is_available": is_available,
                "current_status": item.get("status", {}).get("name"),
            }

        @self.register_tool
        async def create_checkout(
            ctx: RunContext[CirculationDependencies],
            patron_id: str,
            item_id: str,
            service_point_id: str,
            due_date: Optional[str] = None
        ) -> Dict[str, Any]:
            """
            Create a checkout/loan in FOLIO.

            Args:
                patron_id: User UUID
                item_id: Item UUID
                service_point_id: Service point UUID
                due_date: Optional due date (ISO format)

            Returns:
                Created loan record
            """
            client = ctx.deps.folio_client

            # Prepare checkout request
            checkout_data = {
                "itemBarcode": item_id,  # FOLIO accepts barcode or ID
                "userBarcode": patron_id,
                "servicePointId": service_point_id or ctx.deps.service_point_id,
                "loanDate": datetime.utcnow().isoformat(),
            }

            if due_date:
                checkout_data["dueDate"] = due_date

            # Call FOLIO check-out endpoint
            try:
                loan = client.folio_post(
                    "/circulation/check-out-by-barcode",
                    checkout_data
                )
                return {"success": True, "loan": loan}
            except Exception as e:
                return {"success": False, "error": str(e)}

        @self.register_tool
        async def get_circulation_policy(
            ctx: RunContext[CirculationDependencies],
            patron_type: str,
            item_type: str,
            loan_type: str
        ) -> Dict[str, Any]:
            """
            Retrieve applicable circulation policy.

            Args:
                patron_type: Patron group type
                item_type: Material type
                loan_type: Loan type (regular, reserve, etc.)

            Returns:
                Circulation policy with loan period
            """
            # Simplified - in production would use circulation rules engine
            default_policies = {
                "regular": {"loan_period": 21, "renewable": True, "max_renewals": 3},
                "reserve": {"loan_period": 2, "renewable": False, "max_renewals": 0},
                "reference": {"loan_period": 0, "renewable": False, "max_renewals": 0},
            }

            return default_policies.get(loan_type, default_policies["regular"])


# Create singleton instance
checkout_agent = CheckoutAgent()

Step 3: Create Circulation Models (src/edge_ai/circulation/models/loan.py)

"""
Circulation domain models.
"""
from pydantic import BaseModel, Field, UUID4
from typing import Optional, List
from datetime import datetime
from enum import Enum


class LoanStatus(str, Enum):
    """Loan status values."""
    OPEN = "Open"
    CLOSED = "Closed"


class CheckoutRequest(BaseModel):
    """Request to check out an item to a patron."""

    patron_barcode: str = Field(
        description="Patron barcode or ID"
    )
    item_barcode: str = Field(
        description="Item barcode or ID"
    )
    service_point_id: Optional[str] = Field(
        default=None,
        description="Service point where checkout occurs"
    )
    override_blocks: bool = Field(
        default=False,
        description="Whether to override patron blocks (requires permission)"
    )
    due_date_override: Optional[datetime] = Field(
        default=None,
        description="Optional due date override"
    )


class CheckoutResponse(BaseModel):
    """Response from checkout operation."""

    success: bool = Field(
        description="Whether checkout succeeded"
    )
    loan: Optional[Dict] = Field(
        default=None,
        description="Created loan record"
    )
    error_message: Optional[str] = Field(
        default=None,
        description="Error message if checkout failed"
    )
    warnings: List[str] = Field(
        default_factory=list,
        description="Warning messages (e.g., item nearly overdue)"
    )
    patron_blocks: List[str] = Field(
        default_factory=list,
        description="Patron blocks that prevented checkout"
    )
    suggested_action: Optional[str] = Field(
        default=None,
        description="AI-generated suggestion for staff"
    )


class Loan(BaseModel):
    """FOLIO Loan record."""

    id: UUID4 = Field(description="Loan UUID")
    userId: UUID4 = Field(description="Patron UUID")
    itemId: UUID4 = Field(description="Item UUID")
    status: LoanStatus = Field(description="Loan status")
    loanDate: datetime = Field(description="Checkout date/time")
    dueDate: datetime = Field(description="Due date/time")
    returnDate: Optional[datetime] = Field(
        default=None,
        description="Return date/time (if returned)"
    )
    renewalCount: int = Field(
        default=0,
        description="Number of renewals"
    )
    action: str = Field(description="Last action performed")
    itemStatus: str = Field(description="Item status name")

    class Config:
        json_schema_extra = {
            "example": {
                "id": "0e8e7e81-68b9-4af7-b3e5-8c3e6b38c8d2",
                "userId": "79dffc9c-de1b-11e7-9296-cec278b6b50a",
                "itemId": "459afaba-5b39-468d-9072-eb1685e0ddf4",
                "status": "Open",
                "loanDate": "2024-01-15T10:30:00.000Z",
                "dueDate": "2024-02-05T23:59:59.000Z",
                "renewalCount": 0,
                "action": "checkedout",
                "itemStatus": "Checked out"
            }
        }

Step 4: Create Circulation Router (src/edge_ai/circulation/router.py)

"""
Circulation API Router
Exposes circulation workflows as FastAPI endpoints (auto-converted to MCP tools).
"""
from fastapi import APIRouter, HTTPException, Depends
from typing import Optional

from edge_ai.circulation.agents.checkout_agent import checkout_agent, CirculationDependencies
from edge_ai.circulation.models.loan import CheckoutRequest, CheckoutResponse


router = APIRouter(
    prefix="/circulation",
    tags=["circulation"],
)


@router.post(
    "/checkout",
    operation_id="circulation_checkout",
    summary="Check out item to patron",
    description="""
    AI-assisted patron checkout workflow.

    Features:
    - Automatic patron eligibility validation
    - Item availability checking
    - Circulation policy application
    - Helpful error messages for staff

    The AI agent will:
    1. Validate the patron account
    2. Check for patron blocks
    3. Verify item availability
    4. Apply circulation policies
    5. Create the loan record
    6. Provide guidance if checkout cannot complete
    """,
    response_model=CheckoutResponse,
)
async def checkout_item(
    request: CheckoutRequest,
    service_point_id: Optional[str] = None,
) -> CheckoutResponse:
    """
    Check out an item to a patron with AI assistance.

    Args:
        request: Checkout request with patron and item barcodes
        service_point_id: Service point where checkout occurs

    Returns:
        Checkout response with loan record or error details
    """
    # Prepare agent dependencies
    deps = CirculationDependencies(
        service_point_id=service_point_id
    )

    # Build prompt for AI agent
    prompt = f"""
    Check out item with barcode '{request.item_barcode}' to patron with barcode '{request.patron_barcode}'.

    Service point: {service_point_id or 'default'}
    Override blocks: {request.override_blocks}
    Due date override: {request.due_date_override or 'none - use policy'}

    Steps to complete:
    1. Look up patron by barcode and check for blocks
    2. Look up item by barcode and verify availability
    3. Get applicable circulation policy
    4. If all checks pass, create the checkout
    5. Provide helpful feedback about the checkout or why it cannot complete
    """

    try:
        # Run agent
        result = await checkout_agent.run(prompt, deps=deps)
        return result

    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"Checkout failed: {str(e)}"
        )


@router.post(
    "/checkin",
    operation_id="circulation_checkin",
    summary="Check in returned item",
    description="AI-assisted item check-in workflow",
)
async def checkin_item(
    item_barcode: str,
    service_point_id: Optional[str] = None,
):
    """Check in a returned item."""
    # Similar implementation to checkout
    # Would use a checkin_agent
    pass


@router.post(
    "/renew",
    operation_id="circulation_renew_loan",
    summary="Renew a loan",
    description="AI-assisted loan renewal workflow",
)
async def renew_loan(
    item_barcode: str,
    patron_barcode: Optional[str] = None,
):
    """Renew a loan."""
    # Would use a renewal_agent
    pass


@router.get(
    "/loans/overdue",
    operation_id="circulation_get_overdue_loans",
    summary="Get overdue loans",
    description="Retrieve all overdue loans with AI analysis",
)
async def get_overdue_loans(
    patron_barcode: Optional[str] = None,
    limit: int = 100,
):
    """Get overdue loans, optionally filtered by patron."""
    # Would return loans with AI-generated suggestions
    pass

Step 5: Implement Additional Domains

Following the same pattern, implement:

  1. Acquisitions Module

    • order_agent.py - Purchase order creation/management
    • receiving_agent.py - Receiving workflows
    • vendor_agent.py - Vendor management
    • Operations: create_order, receive_items, manage_vendor

  2. Finance Module (extend existing)

    • invoice_agent.py - Already exists, enhance with tools
    • payment_agent.py - Payment approvals
    • budget_agent.py - Budget management
    • Operations: process_invoice, approve_payment, check_budget

  3. Users Module

    • patron_agent.py - Patron account management
    • permission_agent.py - Permission workflows
    • Operations: create_patron, update_permissions, reset_password

  4. Inventory Module (already exists)

    • Enhance existing instance_agent.py
    • Add holdings_agent.py and items_agent.py
    • Operations: catalog_item, update_holdings, transfer_item

Step 6: Update Main App (src/edge_ai/main.py)

# Add to existing imports
from edge_ai.circulation.router import router as circulation_router
from edge_ai.acquisitions.router import router as acquisitions_router
from edge_ai.finance.router import router as finance_router
from edge_ai.users.router import router as users_router

# Add after existing routers
app.include_router(circulation_router)
app.include_router(acquisitions_router)
app.include_router(finance_router)
app.include_router(users_router)

# Configure MCP to expose domain workflows
from fastapi_mcp import FastApiMCP

mcp = FastApiMCP(
    app,
    include_tags=[
        "circulation",
        "acquisitions",
        "finance",
        "users",
        "inventory",  # existing
    ],
)
mcp.mount_http()

Implementation Steps

Phase 1: Foundation (Week 1)

  1. Day 1-2: Shared Infrastructure

    • Create shared/agent_base.py with BaseFOLIOAgent
    • Create shared/folio_client_wrapper.py with enhanced client
    • Create shared/validation.py with common validators
    • Write tests for shared components

  2. Day 3-5: First Domain (Circulation)

    • Create circulation module structure
    • Implement CheckoutAgent with tools
    • Create Loan and Patron models
    • Implement circulation router
    • Write comprehensive tests
    • Test MCP tool exposure

Phase 2: Additional Domains (Weeks 2-6)

  1. Week 2: Acquisitions Module

    • Implement OrderAgent
    • Create Order and Vendor models
    • Implement acquisitions router
    • Write tests

  2. Week 3: Finance Module

    • Enhance existing InvoiceAgent
    • Implement PaymentAgent
    • Create comprehensive finance models
    • Implement finance router
    • Write tests

  3. Week 4: Users Module

    • Implement PatronAgent
    • Create User and Permission models
    • Implement users router
    • Write tests

  4. Week 5: Inventory Enhancement

    • Enhance existing InstanceAgent
    • Implement HoldingsAgent and ItemsAgent
    • Expand inventory models
    • Write additional tests

  5. Week 6: Additional Domains

    • Implement 2-3 additional domain modules based on priority
    • Examples: Organizations, Courses, ERM

Phase 3: Polish & Documentation (Week 7-8)

  1. Week 7: Integration & Testing

    • End-to-end integration tests
    • Performance testing
    • Error handling review
    • Security audit

  2. Week 8: Documentation & Deployment

    • API documentation
    • Agent behavior documentation
    • Deployment guide
    • User training materials

Testing Strategy

Agent Testing Pattern

"""
Test circulation checkout agent.
"""
import pytest
from pydantic_ai.messages import ModelMessage
from pydantic_ai.testing import TestModel

from edge_ai.circulation.agents.checkout_agent import checkout_agent, CirculationDependencies


@pytest.mark.asyncio
async def test_checkout_agent_success(mock_folio_client):
    """Test successful checkout workflow."""

    # Setup dependencies with mock client
    deps = CirculationDependencies(
        folio_client=mock_folio_client,
        service_point_id="test-sp-123"
    )

    # Use TestModel to avoid actual AI API calls
    checkout_agent.agent.model = TestModel()

    # Run agent
    result = await checkout_agent.run(
        prompt="Check out item 123456 to patron 789012",
        deps=deps
    )

    # Assert results
    assert result.success is True
    assert result.loan is not None
    assert len(result.patron_blocks) == 0


@pytest.mark.asyncio
async def test_checkout_agent_patron_blocked(mock_folio_client):
    """Test checkout with patron blocks."""

    # Configure mock to return patron with blocks
    mock_folio_client.configure_patron_blocks(["overdue_fines"])

    deps = CirculationDependencies(folio_client=mock_folio_client)
    checkout_agent.agent.model = TestModel()

    result = await checkout_agent.run(
        prompt="Check out item 123456 to patron 789012",
        deps=deps
    )

    # Should fail with block information
    assert result.success is False
    assert len(result.patron_blocks) > 0
    assert "overdue_fines" in result.patron_blocks[0]
    assert result.suggested_action is not None

Integration Testing

# Test full workflow
pytest tests/circulation/test_checkout_workflow.py -v

# Test all domains
pytest tests/ --cov=src/edge_ai -v

# Test MCP tool exposure
pytest tests/test_mcp_tools.py -v

Effort Estimate

Module Tasks Time Complexity
Shared infrastructure Base classes, wrappers 2 days Medium
Circulation 3 agents, models, router 1 week High
Acquisitions 3 agents, models, router 1 week High
Finance 2 agents, models, router 1 week High
Users 2 agents, models, router 5 days Medium
Inventory (enhance) 2 agents, expand models 5 days Medium
Additional domains 2-3 modules 1 week Medium
Testing & QA Comprehensive tests 1 week Medium
Documentation Docs, guides 3 days Low
Total All domains 8-10 weeks High

Pros & Cons Summary

Pros ✅

  • Semantic operations: Meaningful workflows for AI agents
  • AI assistance: Intelligent validation, suggestions, error handling
  • Production ready: Comprehensive error handling and testing
  • Follows patterns: Extends existing agent architecture
  • Maintainable: Clear domain separation
  • Curated tools: 50-100 meaningful operations vs 1,749 endpoints
  • Business logic: Can encode complex library workflows
  • Better UX: Helpful messages and guidance for staff

Cons ❌

  • High effort: 8-10 weeks of development
  • Limited coverage: Not all 1,749 endpoints exposed
  • Maintenance burden: Need to update when FOLIO APIs change
  • Model complexity: Extensive Pydantic models required
  • Domain expertise: Requires deep FOLIO knowledge
  • Testing complexity: Comprehensive tests needed
  • Documentation: Requires extensive documentation

Option 3: Hybrid Tiered Pattern

Overview

Combines both approaches: Tier 1 provides read-only proxy access to all FOLIO endpoints, Tier 2 offers curated agent-assisted workflows for write operations, and Tier 3 enables custom AI-powered integrations.

Architecture Diagram

┌─────────────┐
│  AI Agent   │
│  (Claude)   │
└──────┬──────┘
       │ MCP Protocol
       │
       ├─ "Get all instances" ────────────┐
       ├─ "Check out book to patron" ─────┤
       └─ "Bulk import MARC records" ─────┤
                                          │
┌─────────────────────────────────────────▼──────┐
│  FastAPI-MCP Server (edge-ai)                  │
│                                                 │
│  ┌───────────────────────────────────────────┐ │
│  │  TIER 1: Read-Only Proxy                 │ │
│  │  GET /api/raw/{module}/{endpoint}        │ │
│  │  • All 1,749 FOLIO endpoints            │ │
│  │  • GET only (safe exploration)          │ │
│  │  • No validation, direct passthrough    │ │
│  │  • Tag: "folio-read"                    │ │
│  └───────────────────────────────────────────┘ │
│                                                 │
│  ┌───────────────────────────────────────────┐ │
│  │  TIER 2: Agent-Assisted Workflows        │ │
│  │  POST /api/workflows/{domain}/{operation}│ │
│  │  • 30-50 curated operations             │ │
│  │  • AI validation & assistance           │ │
│  │  • Pydantic AI agents                   │ │
│  │  • Tag: "workflows"                     │ │
│  │  Examples:                               │ │
│  │  ├─ /workflows/circulation/checkout     │ │
│  │  ├─ /workflows/acquisitions/create_po   │ │
│  │  └─ /workflows/finance/process_invoice  │ │
│  └───────────────────────────────────────────┘ │
│                                                 │
│  ┌───────────────────────────────────────────┐ │
│  │  TIER 3: AI-Powered Integrations         │ │
│  │  POST /api/integrations/{operation}      │ │
│  │  • 5-10 custom AI operations            │ │
│  │  • Multi-step orchestration             │ │
│  │  • Advanced AI capabilities             │ │
│  │  • Tag: "ai-integrations"               │ │
│  │  Examples:                               │ │
│  │  ├─ /integrations/bulk_import_marc      │ │
│  │  ├─ /integrations/intelligent_search    │ │
│  │  └─ /integrations/data_migration        │ │
│  └───────────────────────────────────────────┘ │
└─────────────────────────────────────────────────┘
                       │
                       ▼
                ┌──────────────┐
                │ FOLIO Okapi  │
                │   Gateway    │
                └──────────────┘

File Structure

src/edge_ai/
├── main.py                                    # Update to include all tiers
├── tier1_proxy/                               # Tier 1: Read-only proxy
│   ├── __init__.py
│   ├── router.py                              # GET-only proxy router
│   ├── models.py                              # Generic response models
│   └── catalog.py                             # Endpoint discovery
├── tier2_workflows/                           # Tier 2: Agent workflows
│   ├── __init__.py
│   ├── circulation/
│   │   ├── router.py                          # Circulation workflows
│   │   ├── checkout_agent.py
│   │   └── models.py
│   ├── acquisitions/
│   │   ├── router.py                          # Acquisitions workflows
│   │   ├── order_agent.py
│   │   └── models.py
│   ├── finance/
│   │   ├── router.py                          # Finance workflows
│   │   ├── invoice_agent.py
│   │   └── models.py
│   └── shared/
│       ├── agent_base.py                      # Shared agent base
│       └── validation.py                      # Common validators
├── tier3_integrations/                        # Tier 3: AI integrations
│   ├── __init__.py
│   ├── router.py                              # Integration endpoints
│   ├── bulk_import_agent.py                   # Bulk operations
│   ├── intelligent_search_agent.py            # Advanced search
│   ├── data_migration_agent.py                # Migration workflows
│   └── models.py                              # Integration models
├── shared/
│   ├── __init__.py
│   ├── folio_client_wrapper.py                # Enhanced FOLIO client
│   ├── middleware.py                          # Shared middleware
│   └── utils.py                               # Utility functions
└── tests/
    ├── tier1/
    ├── tier2/
    └── tier3/

Detailed Implementation

Tier 1: Read-Only Proxy (src/edge_ai/tier1_proxy/router.py)

"""
Tier 1: Read-Only FOLIO API Proxy
Provides safe read access to all 1,749 FOLIO endpoints.
"""
from fastapi import APIRouter, Request, HTTPException, Query
from typing import Dict, Any, Optional
import httpx
import os

router = APIRouter(
    prefix="/api/raw",
    tags=["folio-read"],
)


@router.get(
    "/{module}/{path:path}",
    operation_id="folio_read_{module}_{path}",
    summary="Read FOLIO data (GET only)",
    description="""
    **Tier 1: Read-Only Proxy**

    Safe exploration of FOLIO data. All 1,749 endpoints accessible via GET.

    **Safety**: Read-only access prevents accidental modifications.
    **Coverage**: Complete access to all FOLIO data.
    **Use Case**: Data exploration, reporting, analysis.

    For write operations, use Tier 2 workflows (POST /api/workflows/...).
    """,
)
async def read_folio_data(
    module: str,
    path: str,
    request: Request,
    query: Optional[str] = Query(None, description="CQL query string"),
    limit: int = Query(10, ge=1, le=1000, description="Max results"),
    offset: int = Query(0, ge=0, description="Pagination offset"),
) -> Dict[str, Any]:
    """
    Read data from FOLIO API (GET only).

    Args:
        module: FOLIO module name
        path: Endpoint path
        query: Optional CQL query
        limit: Maximum results
        offset: Pagination offset

    Returns:
        FOLIO API response
    """
    okapi_url = os.getenv("OKAPI_URL")
    tenant_id = os.getenv("TENANT_ID")

    # Build URL
    endpoint = f"/{module}/{path}"
    full_url = f"{okapi_url.rstrip('/')}{endpoint}"

    # Build query parameters
    params = dict(request.query_params)
    if query:
        params["query"] = query
    if "limit" not in params:
        params["limit"] = limit
    if "offset" not in params:
        params["offset"] = offset

    # Get auth headers
    headers = {
        "X-Okapi-Tenant": tenant_id,
        "Accept": "application/json",
    }

    # Make request
    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            response = await client.get(
                url=full_url,
                params=params,
                headers=headers,
            )

            if response.status_code >= 400:
                raise HTTPException(
                    status_code=response.status_code,
                    detail={
                        "tier": 1,
                        "message": "Read request failed",
                        "endpoint": endpoint,
                        "folio_response": response.text,
                    }
                )

            return response.json() if response.content else {}

        except httpx.RequestError as e:
            raise HTTPException(
                status_code=503,
                detail=f"FOLIO connection failed: {str(e)}"
            )


@router.get(
    "/catalog",
    operation_id="folio_catalog",
    summary="Discover available FOLIO endpoints",
    description="Get a catalog of available FOLIO API endpoints",
)
async def get_endpoint_catalog(
    module: Optional[str] = None,
) -> Dict[str, Any]:
    """
    Get catalog of FOLIO endpoints.

    Returns:
        Dictionary of modules and their endpoints
    """
    # This would ideally scrape FOLIO's API documentation
    # or use module descriptors to build a catalog
    # For now, return curated list

    catalog = {
        "inventory-storage": [
            "instances",
            "holdings",
            "items",
            "instance-types",
            "contributor-types",
        ],
        "circulation": [
            "loans",
            "requests",
            "check-out-by-barcode",
            "check-in-by-barcode",
        ],
        "acquisitions": [
            "composite-orders",
            "orders",
            "order-lines",
        ],
        "finance": [
            "invoices",
            "vouchers",
            "transactions",
        ],
        "users": [
            "users",
            "groups",
            "permissions",
        ],
    }

    if module:
        return {module: catalog.get(module, [])}

    return catalog

Tier 2: Workflows Router (src/edge_ai/tier2_workflows/circulation/router.py)

"""
Tier 2: Circulation Workflows
Agent-assisted circulation operations with validation and error handling.
"""
from fastapi import APIRouter, HTTPException
from typing import Optional

from edge_ai.tier2_workflows.circulation.checkout_agent import checkout_agent
from edge_ai.tier2_workflows.shared.agent_base import BaseFOLIODependencies

router = APIRouter(
    prefix="/api/workflows/circulation",
    tags=["workflows"],
)


@router.post(
    "/checkout",
    operation_id="workflow_circulation_checkout",
    summary="[Workflow] Check out item to patron",
    description="""
    **Tier 2: Agent-Assisted Workflow**

    Check out an item to a patron with AI assistance.

    **Features**:
    - Automatic patron validation
    - Item availability checking
    - Circulation policy application
    - Intelligent error messages
    - Suggested actions for staff

    **Safety**: Validated write operation with comprehensive checks.

    For read-only access, use Tier 1: GET /api/raw/circulation/loans
    """,
)
async def checkout(
    patron_barcode: str,
    item_barcode: str,
    service_point_id: Optional[str] = None,
    override_blocks: bool = False,
):
    """Execute AI-assisted checkout workflow."""

    deps = BaseFOLIODependencies()

    prompt = f"""
    Check out item '{item_barcode}' to patron '{patron_barcode}'.
    Override blocks: {override_blocks}
    Service point: {service_point_id or 'default'}
    """

    try:
        result = await checkout_agent.run(prompt, deps=deps)
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


# Additional workflow endpoints...

Tier 3: AI Integrations (src/edge_ai/tier3_integrations/router.py)

"""
Tier 3: AI-Powered Integrations
Advanced AI operations for complex workflows.
"""
from fastapi import APIRouter, HTTPException, UploadFile, File
from typing import List, Dict, Any
from pydantic import BaseModel

from edge_ai.tier3_integrations.bulk_import_agent import bulk_import_agent
from edge_ai.tier3_integrations.intelligent_search_agent import search_agent

router = APIRouter(
    prefix="/api/integrations",
    tags=["ai-integrations"],
)


class BulkImportRequest(BaseModel):
    """Request for bulk import operation."""
    data_type: str  # "marc", "csv", "json"
    target_module: str  # "inventory", "acquisitions", etc.
    validation_mode: str = "strict"  # "strict", "lenient", "auto-fix"
    dry_run: bool = False


class BulkImportResponse(BaseModel):
    """Response from bulk import."""
    success: bool
    total_records: int
    imported: int
    failed: int
    errors: List[Dict[str, Any]]
    suggestions: List[str]


@router.post(
    "/bulk-import",
    operation_id="integration_bulk_import",
    summary="[AI Integration] Bulk import records",
    description="""
    **Tier 3: AI-Powered Integration**

    Intelligent bulk import of records with AI assistance.

    **AI Features**:
    - Automatic format detection
    - Data validation and cleaning
    - Error auto-correction
    - Duplicate detection
    - Progress reporting
    - Rollback on failure

    **Supports**: MARC, CSV, JSON formats
    **Use Case**: Data migrations, batch cataloging

    This is an advanced operation that orchestrates multiple FOLIO API calls.
    """,
    response_model=BulkImportResponse,
)
async def bulk_import(
    request: BulkImportRequest,
    file: UploadFile = File(...),
) -> BulkImportResponse:
    """
    Bulk import records with AI assistance.

    Args:
        request: Import configuration
        file: Data file (MARC/CSV/JSON)

    Returns:
        Import results with statistics and error details
    """
    # Read file
    content = await file.read()

    # Build prompt
    prompt = f"""
    Import {request.data_type} records into {request.target_module}.
    Validation mode: {request.validation_mode}
    Dry run: {request.dry_run}

    File size: {len(content)} bytes

    Tasks:
    1. Parse and validate records
    2. Check for duplicates
    3. Auto-fix common errors (if validation_mode allows)
    4. {'Simulate' if request.dry_run else 'Execute'} import
    5. Report results with detailed error information
    """

    try:
        result = await bulk_import_agent.run(
            prompt,
            file_content=content,
            config=request,
        )
        return result
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"Bulk import failed: {str(e)}"
        )


@router.post(
    "/intelligent-search",
    operation_id="integration_intelligent_search",
    summary="[AI Integration] Intelligent search",
    description="""
    **Tier 3: AI-Powered Integration**

    Natural language search across FOLIO with AI understanding.

    **AI Features**:
    - Natural language query understanding
    - Multi-module searching
    - Relevance ranking
    - Result summarization
    - Related record suggestions

    **Example**: "Find all books about Shakespeare published after 2000"

    For simple searches, use Tier 1: GET /api/raw/inventory-storage/instances?query=...
    """,
)
async def intelligent_search(
    query: str,
    modules: List[str] = None,
    limit: int = 10,
):
    """
    Search FOLIO with natural language understanding.

    Args:
        query: Natural language search query
        modules: Optional module filter
        limit: Maximum results

    Returns:
        Intelligent search results with AI analysis
    """
    prompt = f"""
    Search query: "{query}"
    Modules: {modules or 'all'}
    Limit: {limit}

    Tasks:
    1. Parse natural language query into CQL
    2. Determine which FOLIO modules to search
    3. Execute searches across modules
    4. Rank results by relevance
    5. Summarize findings
    6. Suggest related searches
    """

    try:
        result = await search_agent.run(prompt, limit=limit)
        return result
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"Search failed: {str(e)}"
        )


@router.post(
    "/data-migration",
    operation_id="integration_data_migration",
    summary="[AI Integration] Data migration assistant",
    description="""
    **Tier 3: AI-Powered Integration**

    AI-assisted data migration between systems or FOLIO tenants.

    **AI Features**:
    - Mapping schema differences
    - Data transformation
    - Dependency resolution
    - Progress tracking
    - Rollback capability
    """,
)
async def data_migration(
    source: str,
    target: str,
    migration_type: str,
    mapping_rules: Dict[str, Any] = None,
):
    """Execute AI-assisted data migration."""
    # Implementation similar to bulk_import
    pass

Main App Integration (src/edge_ai/main.py)

"""
Main FastAPI application with three-tier MCP architecture.
"""
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi_mcp import FastApiMCP

# Import tier routers
from edge_ai.tier1_proxy.router import router as tier1_router
from edge_ai.tier2_workflows.circulation.router import router as circulation_router
from edge_ai.tier2_workflows.acquisitions.router import router as acquisitions_router
from edge_ai.tier2_workflows.finance.router import router as finance_router
from edge_ai.tier3_integrations.router import router as tier3_router

# Existing routers
from edge_ai.inventory.router import router as inventory_router

app = FastAPI(
    title="FOLIO Edge AI - Hybrid MCP Server",
    version="0.5.0",
    description="""
    Multi-tier MCP server for FOLIO library system.

    **Tier 1: Read-Only Proxy** (`/api/raw/*`)
    - All 1,749 FOLIO endpoints (GET only)
    - Safe data exploration
    - No validation, direct passthrough

    **Tier 2: Agent-Assisted Workflows** (`/api/workflows/*`)
    - 30-50 curated operations
    - AI validation and assistance
    - Production-ready workflows

    **Tier 3: AI-Powered Integrations** (`/api/integrations/*`)
    - 5-10 advanced operations
    - Multi-step AI orchestration
    - Complex workflows
    """,
)

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Register routers
app.include_router(tier1_router)         # Tier 1: Read proxy
app.include_router(circulation_router)   # Tier 2: Workflows
app.include_router(acquisitions_router)  # Tier 2: Workflows
app.include_router(finance_router)       # Tier 2: Workflows
app.include_router(tier3_router)         # Tier 3: AI integrations
app.include_router(inventory_router)     # Existing

# Configure MCP with tier-based filtering
mcp = FastApiMCP(
    app,
    # Expose all tiers, but allow filtering via tags
    include_tags=[
        "folio-read",        # Tier 1
        "workflows",         # Tier 2
        "ai-integrations",   # Tier 3
        "inventory",         # Existing
    ],
)

# Use HTTP transport (recommended)
mcp.mount_http()


@app.get("/")
async def root():
    """Service info endpoint."""
    return {
        "service": "FOLIO Edge AI - Hybrid MCP Server",
        "version": "0.5.0",
        "tiers": {
            "tier1": {
                "name": "Read-Only Proxy",
                "prefix": "/api/raw",
                "description": "Safe read access to all FOLIO endpoints",
                "tag": "folio-read",
            },
            "tier2": {
                "name": "Agent-Assisted Workflows",
                "prefix": "/api/workflows",
                "description": "Curated operations with AI validation",
                "tag": "workflows",
            },
            "tier3": {
                "name": "AI-Powered Integrations",
                "prefix": "/api/integrations",
                "description": "Advanced AI orchestration",
                "tag": "ai-integrations",
            },
        },
        "mcp_endpoint": "/mcp",
    }

Implementation Steps

Phase 1: Tier 1 Implementation (Week 1)

  1. Days 1-2: Core Proxy

    • Create tier1_proxy/router.py with GET-only proxy
    • Implement endpoint catalog/discovery
    • Add tests for read operations
    • Configure MCP exposure

  2. Day 3: Enhancement

    • Add query parameter handling (CQL, pagination)
    • Implement response caching
    • Add request logging
    • Document Tier 1 usage

  3. Days 4-5: Testing & Polish

    • Comprehensive testing against FOLIO
    • Performance testing
    • Documentation
    • Deploy Tier 1 to staging

Phase 2: Tier 2 Implementation (Weeks 2-5)

  1. Week 2: Foundation

    • Create tier2_workflows/shared/ infrastructure
    • Implement base agent class
    • Create first workflow (circulation checkout)
    • Test workflow agent

  2. Weeks 3-4: Core Workflows

    • Implement 3-5 circulation workflows
    • Implement 3-5 acquisitions workflows
    • Implement 3-5 finance workflows
    • Create comprehensive tests

  3. Week 5: Additional Workflows

    • Implement 2-3 workflows per domain (users, inventory, etc.)
    • Total: 30-40 workflows
    • Integration testing
    • Documentation

Phase 3: Tier 3 Implementation (Weeks 6-7)

  1. Week 6: Core Integrations

    • Implement bulk import agent
    • Implement intelligent search agent
    • Create integration tests
    • Performance testing

  2. Week 7: Additional Integrations

    • Implement data migration agent
    • Implement 2-3 additional AI integrations
    • Total: 5-8 integrations
    • Integration testing

Phase 4: Polish & Deploy (Week 8)

  1. Week 8: Finalization
    • End-to-end testing
    • Security audit
    • Performance optimization
    • Comprehensive documentation
    • Deployment to production

Testing Strategy

Tier 1 Testing

# Test read-only proxy
async def test_tier1_read_instances():
    response = client.get("/api/raw/inventory-storage/instances?limit=5")
    assert response.status_code == 200
    assert "instances" in response.json()

# Test write prevention
async def test_tier1_no_write():
    response = client.post("/api/raw/inventory-storage/instances", json={})
    assert response.status_code == 405  # Method not allowed

Tier 2 Testing

# Test workflow agent
async def test_tier2_checkout_workflow():
    response = client.post(
        "/api/workflows/circulation/checkout",
        json={
            "patron_barcode": "123",
            "item_barcode": "456",
        }
    )
    assert response.status_code == 200
    assert response.json()["success"] is True

Tier 3 Testing

# Test AI integration
async def test_tier3_bulk_import():
    with open("test_data.marc", "rb") as f:
        response = client.post(
            "/api/integrations/bulk-import",
            data={"data_type": "marc", "target_module": "inventory"},
            files={"file": f}
        )
    assert response.status_code == 200
    assert response.json()["imported"] > 0

Effort Estimate

Phase Tasks Time Complexity
Tier 1 Read-only proxy, catalog 1 week Low
Tier 2 30-40 workflows, 5-8 agents 4 weeks High
Tier 3 5-8 AI integrations 2 weeks High
Testing & Polish E2E tests, docs 1 week Medium
Total All tiers 8 weeks Medium-High

Incremental Deployment Timeline

  • Week 1: Deploy Tier 1 (immediate value)
  • Week 5: Deploy Tier 2 with core workflows
  • Week 7: Deploy Tier 3 with AI integrations
  • Week 8: Production release

Pros & Cons Summary

Pros ✅

  • Flexibility: Supports both exploration (Tier 1) and production workflows (Tier 2/3)
  • Safety: Read operations separate from writes
  • Incremental: Can deploy tiers independently
  • Coverage: All endpoints accessible via Tier 1
  • Intelligence: AI assistance where it matters (Tier 2/3)
  • Manageable tools: ~50-80 total MCP tools (not 1,749)
  • Progressive: Start simple, add complexity as needed
  • Best of both worlds: Raw access + curated workflows

Cons ❌

  • Most complex: Three tiers to maintain
  • Learning curve: Users need to understand tier system
  • Documentation: Requires clear tier usage guidelines
  • Overlap: Some functionality in multiple tiers
  • Coordination: Need to ensure tier consistency

Comparison Matrix

Criteria Option 1: Proxy Option 2: Agents Option 3: Hybrid
Implementation Time 2 days 8-10 weeks 8 weeks
Coverage 100% (1,749 endpoints) 30% (~500 operations) 100% read + 30% write
MCP Tools Created 1,749 50-100 60-100
AI Assistance None High Medium-High
Safety Low High Medium-High
Flexibility Maximum Limited High
Maintenance Low High Medium
Production Ready Low High Medium-High
Learning Curve Low Medium Medium
Incremental Deploy No No Yes (tier by tier)
Testing Complexity Low High Medium
Documentation Needs Low High Medium-High

Recommendation

I recommend Option 3 (Hybrid Tiered Pattern) because it:

  1. Provides immediate value: Tier 1 can be deployed in Week 1
  2. Balances coverage and usability: All endpoints accessible, but curated workflows for common operations
  3. Supports iteration: Deploy tiers independently as they're ready
  4. Manages tool count: ~60-100 meaningful tools instead of 1,749
  5. Combines strengths: Raw access for exploration + AI assistance for production workflows
  6. Enables innovation: Tier 3 allows advanced AI-powered features
  7. Progressive complexity: Start with Tier 1, add Tier 2/3 as needed

Quick Start Path:

  • Week 1: Deploy Tier 1 (read-only proxy) - immediate access to all FOLIO data
  • Weeks 2-5: Build Tier 2 workflows based on actual usage patterns
  • Weeks 6-7: Add Tier 3 integrations for advanced use cases

This approach minimizes risk while maximizing value delivery.

Clone this wiki locally