finizi-mcp by hqtrung - MCP Server

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)

Clone the repository:

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

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"

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

Copy the environment template:

cp .env.example .env

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)

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"
      }
    }
  }
}

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:

Authentication: User credentials are sent to B4B API via login tool
Token Storage: JWT tokens are stored in MCP session metadata
Automatic Injection: Tokens are automatically added to all API requests
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 for guidelines on:

Code style and standards
Testing requirements
Pull request process
Issue reporting

License

This project is licensed under the MIT License - see 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