Finizi B4B 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) 1. **Clone the repository**: ```bash git clone https://github.com/finizi/finizi-mcp.git cd finizi-mcp ``` 2. **Install UV package manager** (if not already installed): ```bash # 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**: ```bash uv venv source .venv/bin/activate # On Windows: .venv\Scripts\activate uv pip install -e "." ``` ### Installation with pip (Alternative) ```bash python -m venv .venv source .venv/bin/activate # On Windows: .venv\Scripts\activate pip install -e "." ``` ### Configuration 1. **Copy the environment template**: ```bash cp .env.example .env ``` 2. **Configure your environment variables**: ```bash # 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 ```bash python run_server.py ``` #### With MCP Client (Claude Desktop) 1. **Configure Claude Desktop** (`claude_desktop_config.json`): ```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 ```python # 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 ```python # 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 ```python # 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 ```python # 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 ```python # 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 ```python # 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 ```bash # 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 ```bash # 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 ```mermaid 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: ```bash # In .env file LOG_LEVEL=DEBUG ``` View detailed request/response logs: ```python import structlog structlog.configure( processors=[ structlog.dev.ConsoleRenderer(colors=True) ] ) ``` ## Contributing We welcome contributions! Please see [CONTRIBUTING.md](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](LICENSE) file for details. ## Support - **Documentation**: [https://docs.finizi.com/mcp](https://docs.finizi.com/mcp) - **Issues**: [GitHub Issues](https://github.com/finizi/finizi-mcp/issues) - **Email**: dev@finizi.ai - **Discord**: [Finizi Community](https://discord.gg/finizi) ## Acknowledgments - Built with [MCP SDK](https://github.com/anthropics/mcp-python) by Anthropic - HTTP client powered by [httpx](https://www.python-httpx.org/) - Logging with [structlog](https://www.structlog.org/) - Configuration with [Pydantic](https://pydantic-docs.helpmanual.io/) --- **Version**: 1.0.0 **Last Updated**: October 2024 **Maintained by**: Finizi Team