Finizi B4B MCP Server

Overview

The Finizi B4B MCP Server is a Model Context Protocol (MCP) server that provides seamless integration with the Finizi B4B (Business-to-Business) API platform. This server exposes 15 comprehensive MCP tools that enable AI assistants to interact with business entities, invoices, vendors, and products through natural language commands.

Built with modern Python async/await patterns and featuring robust error handling, automatic retries, and token-based authentication, this MCP server provides a reliable bridge between AI assistants and the Finizi B4B platform.

Key Features

15 Comprehensive MCP Tools

The server implements 15 specialized tools across 5 categories:

Authentication Tools (3)

• login: Authenticate users with phone and password
• logout: Clear authentication session
• whoami: Get current user information

Entity Management Tools (4)

• list_entities: List business entities with pagination and search
• get_entity: Retrieve detailed entity information
• create_entity: Create new business entities
• update_entity: Update existing entity details

Invoice Management Tools (4)

• list_invoices: List invoices with advanced filtering
• get_invoice: Retrieve detailed invoice information
• import_invoice_xml: Import invoices from XML format
• get_invoice_statistics: Get invoice analytics and statistics

Vendor Management Tools (2)

• list_vendors: List vendors with pagination and search
• get_vendor: Retrieve detailed vendor information

Product Management Tools (2)

• list_products: List products with category filtering
• search_similar_products: Find similar products using AI matching

Architecture Highlights

• Token Pass-Through Architecture: Secure JWT token management with automatic token extraction and injection
• Singleton API Client: Efficient connection reuse with pooling and keep-alive
• Automatic Retry Logic: Exponential backoff for transient failures (3 attempts)
• Structured Logging: Comprehensive logging with structlog for debugging
• Type Safety: Full type hints and Pydantic models for validation
• Error Handling: Custom exception hierarchy with proper HTTP status mapping

Requirements

• Python 3.11 or higher
• UV package manager (recommended) or pip
• Access to Finizi B4B API server
• Valid B4B API credentials

Quick Start Guide

Installation with UV (Recommended)

1. Clone the repository:

git clone https://github.com/finizi/finizi-mcp.git
cd finizi-mcp

2. Install UV package manager (if not already installed):

# On macOS/Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# On Windows
powershell -c "irm https://astral.sh/uv/install.ps1 | iex"

3. Create virtual environment and install dependencies:

uv venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate
uv pip install -e "."

Installation with pip (Alternative)

python -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate
pip install -e "."

Configuration

1. Copy the environment template:

cp .env.example .env

2. Configure your environment variables:

# Edit .env file
B4B_API_BASE_URL=https://api.finizi.com  # Your B4B API URL
B4B_API_VERSION=v1                       # API version
API_TIMEOUT=30                           # Request timeout in seconds
API_CONNECT_TIMEOUT=10                   # Connection timeout
MAX_RETRIES=3                            # Number of retry attempts
LOG_LEVEL=INFO                           # Logging level

Running the Server

Standalone Mode

python run_server.py

With MCP Client (Claude Desktop)

1. Configure Claude Desktop (claude_desktop_config.json):

{
  "mcpServers": {
    "finizi-b4b": {
      "command": "python",
      "args": ["/path/to/finizi-mcp/run_server.py"],
      "env": {
        "B4B_API_BASE_URL": "https://api.finizi.com"
      }
    }
  }
}

2. Restart Claude Desktop to load the MCP server

Testing the Connection

# In Claude or any MCP client
await login("+84909495665", "YourPassword123@")
# Returns: {"success": true, "message": "Successfully logged in as user@example.com"}

await whoami()
# Returns user information

await list_entities(page=1, per_page=10)
# Returns paginated list of entities

Example Usage

Authentication Flow

# Step 1: Login
result = await login("+84909495665", "SecurePassword123@")
if result["success"]:
    print(f"Logged in as: {result['email']}")

# Step 2: Check current user
user_info = await whoami()
print(f"User ID: {user_info['user']['id']}")

# Step 3: Perform operations
entities = await list_entities(search="Tech Corp")

# Step 4: Logout when done
await logout()

Entity Management

# List all entities with pagination
entities = await list_entities(
    page=1,
    per_page=20,
    search="technology"
)

# Get specific entity details
entity = await get_entity("550e8400-e29b-41d4-a716-446655440000")

# Create new entity
new_entity = await create_entity(
    name="Tech Innovations Ltd",
    entity_type="COMPANY",
    tax_code="1234567890",
    address="123 Tech Street",
    city="San Francisco",
    country="USA"
)

# Update entity
updated = await update_entity(
    entity_id="550e8400-e29b-41d4-a716-446655440000",
    name="Tech Innovations Inc",
    address="456 Innovation Ave"
)

Invoice Management

# List invoices with filters
invoices = await list_invoices(
    entity_id="550e8400-e29b-41d4-a716-446655440000",
    date_from="2024-01-01",
    date_to="2024-12-31",
    status=1,  # ACTIVE
    page=1,
    per_page=50
)

# Get invoice details
invoice = await get_invoice(
    entity_id="550e8400-e29b-41d4-a716-446655440000",
    invoice_id="invoice-uuid-here"
)

# Import XML invoice
xml_content = """<?xml version="1.0"?>
<Invoice>
    <InvoiceNumber>INV-2024-001</InvoiceNumber>
    <Date>2024-01-15</Date>
    <Total>1500.00</Total>
</Invoice>"""
imported = await import_invoice_xml(
    entity_id="550e8400-e29b-41d4-a716-446655440000",
    xml_content=xml_content
)

# Get statistics
stats = await get_invoice_statistics(
    entity_id="550e8400-e29b-41d4-a716-446655440000",
    date_from="2024-01-01",
    date_to="2024-12-31"
)

Vendor Management

# List vendors
vendors = await list_vendors(
    entity_id="550e8400-e29b-41d4-a716-446655440000",
    search="supplies",
    page=1,
    per_page=20
)

# Get vendor details
vendor = await get_vendor(
    entity_id="550e8400-e29b-41d4-a716-446655440000",
    vendor_id="vendor-uuid-here"
)

Product Management

# List products
products = await list_products(
    entity_id="550e8400-e29b-41d4-a716-446655440000",
    category="electronics",
    search="laptop",
    page=1,
    per_page=20
)

# Search similar products
similar = await search_similar_products(
    entity_id="550e8400-e29b-41d4-a716-446655440000",
    query="15-inch laptop with 16GB RAM",
    max_results=10
)

Development Guide

Running Tests

# Run all tests
pytest

# Run with coverage
pytest --cov=src/finizi_b4b_mcp --cov-report=html

# Run specific test module
pytest tests/test_auth.py -v

Code Quality

# Format code
ruff format .

# Lint code
ruff check .

# Fix auto-fixable issues
ruff check --fix .

Project Structure

finizi-mcp/
├── src/
│   └── finizi_b4b_mcp/
│       ├── __init__.py           # Package initialization
│       ├── server.py             # MCP server setup
│       ├── config.py             # Configuration management
│       ├── auth/
│       │   └── token_handler.py  # JWT token management
│       ├── client/
│       │   └── api_client.py     # HTTP client with retry
│       ├── tools/
│       │   ├── auth.py          # Authentication tools
│       │   ├── entities.py      # Entity management tools
│       │   ├── invoices.py      # Invoice management tools
│       │   ├── vendors.py       # Vendor management tools
│       │   └── products.py      # Product management tools
│       └── utils/
│           ├── validators.py    # Input validation
│           ├── formatters.py    # Output formatting
│           └── errors.py        # Custom exceptions
├── tests/
│   ├── test_auth.py             # Authentication tests
│   ├── test_entities.py         # Entity tests
│   └── test_invoices.py         # Invoice tests
├── docs/
│   ├── API_MAPPING.md          # API endpoint mapping
│   └── DEVELOPMENT.md          # Development guide
├── run_server.py                # Server entry point
├── pyproject.toml              # Project configuration
├── .env.example                # Environment template
└── README.md                   # This file

Architecture Overview

Token Pass-Through Model

The server implements a secure token pass-through architecture:

1. Authentication: User credentials are sent to B4B API via login tool
2. Token Storage: JWT tokens are stored in MCP session metadata
3. Automatic Injection: Tokens are automatically added to all API requests
4. Session Isolation: Each MCP session maintains its own authentication state

sequenceDiagram
    participant Client as MCP Client
    participant Server as MCP Server
    participant API as B4B API

    Client->>Server: login(phone, password)
    Server->>API: POST /api/v1/auth/login
    API-->>Server: JWT Token
    Server-->>Client: Success + Store in Session

    Client->>Server: list_entities()
    Server->>Server: Extract Token from Session
    Server->>API: GET /entities [Bearer Token]
    API-->>Server: Entity Data
    Server-->>Client: Formatted Results

Error Handling

The server implements a comprehensive error handling strategy:

• MCPAuthenticationError: 401 Unauthorized - Token invalid/expired
• MCPAuthorizationError: 403 Forbidden - Insufficient permissions
• MCPValidationError: 400 Bad Request - Invalid input parameters
• MCPNotFoundError: 404 Not Found - Resource doesn't exist
• MCPServerError: 500+ Server Error - B4B API issues

Performance Optimizations

• Connection Pooling: Reuse HTTP connections (100 connections, 10 per host)
• Keep-Alive: Maintain persistent connections for reduced latency
• Retry Logic: Automatic retry with exponential backoff (3 attempts)
• Request Timeouts: Configurable timeouts to prevent hanging
• Async Operations: Full async/await support for concurrent operations

Security Considerations

Authentication Security

• JWT tokens are stored only in memory (session metadata)
• Tokens are never logged or persisted to disk
• Each session maintains isolated authentication state
• Automatic token cleanup on logout

API Security

• All API calls use HTTPS in production
• Input validation on all user-provided data
• SQL injection prevention through parameterized queries
• XSS prevention through proper output encoding

Best Practices

• Use strong passwords for B4B accounts
• Rotate API credentials regularly
• Monitor access logs for suspicious activity
• Keep dependencies updated for security patches
• Use environment variables for sensitive configuration

Troubleshooting

Common Issues

Connection Refused

Error: Connection refused to B4B API

Solution: Ensure B4B API server is running and accessible at configured URL

Authentication Failed

Error: Login failed: Invalid credentials

Solution: Verify phone number format (+84...) and password

Token Expired

Error: JWT token expired

Solution: Call login again to obtain a new token

Rate Limiting

Error: Too many requests

Solution: Implement request throttling or increase rate limits

Debug Mode

Enable debug logging for troubleshooting:

# In .env file
LOG_LEVEL=DEBUG

View detailed request/response logs:

import structlog
structlog.configure(
    processors=[
        structlog.dev.ConsoleRenderer(colors=True)
    ]
)

Contributing

We welcome contributions! Please see CONTRIBUTING.md for guidelines on:

• Code style and standards
• Testing requirements
• Pull request process
• Issue reporting

License

This project is licensed under the MIT License - see LICENSE file for details.

Support

• Documentation: https://docs.finizi.com/mcp
• Issues: GitHub Issues
• Email: dev@finizi.ai
• Discord: Finizi Community

Acknowledgments

• Built with MCP SDK by Anthropic
• HTTP client powered by httpx
• Logging with structlog
• Configuration with Pydantic

---

Version: 1.0.0 Last Updated: October 2024 Maintained by: Finizi Team