Finizi B4B MCP Server

# Development Guide ## Table of Contents 1. [Development Environment Setup](#development-environment-setup) 2. [Project Structure](#project-structure) 3. [Code Patterns and Conventions](#code-patterns-and-conventions) 4. [Adding New MCP Tools](#adding-new-mcp-tools) 5. [Testing Guide](#testing-guide) 6. [Debugging Tips](#debugging-tips) 7. [Common Issues and Solutions](#common-issues-and-solutions) 8. [Performance Considerations](#performance-considerations) 9. [Security Best Practices](#security-best-practices) ## Development Environment Setup ### Prerequisites - **Python 3.11+**: Required for async features and type hints - **UV Package Manager**: Fast, reliable Python package management - **Git**: Version control - **IDE**: VSCode or PyCharm recommended ### Initial Setup #### 1. Clone and Navigate ```bash git clone https://github.com/finizi/finizi-mcp.git cd finizi-mcp ``` #### 2. Install UV (if not installed) ```bash # macOS/Linux curl -LsSf https://astral.sh/uv/install.sh | sh # Windows powershell -c "irm https://astral.sh/uv/install.ps1 | iex" # Verify installation uv --version ``` #### 3. Create Virtual Environment ```bash # Using UV (recommended) uv venv source .venv/bin/activate # Windows: .venv\Scripts\activate # Using standard Python python -m venv .venv source .venv/bin/activate ``` #### 4. Install Dependencies ```bash # Production dependencies only uv pip install -e "." # With development dependencies uv pip install -e ".[dev]" ``` #### 5. Environment Configuration ```bash # Copy template cp .env.example .env # Edit configuration vim .env # or use your preferred editor ``` **Key Environment Variables:** ```bash # API Configuration B4B_API_BASE_URL=http://localhost:8000 # B4B API server URL B4B_API_VERSION=v1 # API version # Timeouts (seconds) API_TIMEOUT=30 # Total request timeout API_CONNECT_TIMEOUT=10 # Connection timeout # Retry Configuration MAX_RETRIES=3 # Number of retry attempts RETRY_BACKOFF=1.0 # Backoff multiplier # Logging LOG_LEVEL=INFO # DEBUG|INFO|WARNING|ERROR # Rate Limiting RATE_LIMIT_REQUESTS=100 # Max requests per minute ``` ## Project Structure ### Directory Layout ``` finizi-mcp/ ├── src/ │ └── finizi_b4b_mcp/ # Main package │ ├── __init__.py # Package init & exports │ ├── server.py # MCP server setup │ ├── config.py # Configuration management │ ├── auth/ # Authentication module │ │ ├── __init__.py │ │ └── token_handler.py # JWT token extraction │ ├── client/ # HTTP client module │ │ ├── __init__.py │ │ └── api_client.py # API client with retry │ ├── tools/ # MCP tool implementations │ │ ├── __init__.py │ │ ├── auth.py # Authentication tools │ │ ├── entities.py # Entity CRUD tools │ │ ├── invoices.py # Invoice management │ │ ├── vendors.py # Vendor management │ │ └── products.py # Product management │ └── utils/ # Utility functions │ ├── __init__.py │ ├── validators.py # Input validation │ ├── formatters.py # Output formatting │ └── errors.py # Custom exceptions ├── tests/ # Test suite │ ├── __init__.py │ ├── fixtures/ # Test fixtures │ ├── test_auth.py # Auth tool tests │ ├── test_entities.py # Entity tool tests │ └── test_invoices.py # Invoice tool tests ├── docs/ # Documentation │ ├── API_MAPPING.md # API endpoint mapping │ └── DEVELOPMENT.md # This file ├── run_server.py # Server entry point ├── pyproject.toml # Project configuration ├── uv.lock # Dependency lock file ├── .env.example # Environment template └── README.md # Project documentation ``` ### Module Responsibilities #### `server.py` - MCP server initialization - Tool registration - Lifespan management - Health checks #### `config.py` - Environment variable loading - Configuration validation - Default values #### `auth/token_handler.py` - Token extraction from session - Token validation - Session management #### `client/api_client.py` - HTTP client singleton - Connection pooling - Retry logic - Request/response logging #### `tools/*.py` - MCP tool implementations - Input validation - API calls - Response formatting #### `utils/validators.py` - UUID validation - Pagination validation - Input sanitization #### `utils/formatters.py` - Response formatting - Error message formatting - Data transformation #### `utils/errors.py` - Custom exception classes - Error code mapping - HTTP status handling ## Code Patterns and Conventions ### Code Style #### Python Style Guide - Follow PEP 8 with 100-character line limit - Use type hints for all functions - Docstrings for all public functions - Async/await for I/O operations #### Naming Conventions ```python # Classes: PascalCase class ApiClient: pass # Functions/methods: snake_case async def list_entities(): pass # Constants: UPPER_SNAKE_CASE MAX_RETRIES = 3 # Private methods: leading underscore async def _validate_input(): pass ``` ### Type Hints ```python from typing import Optional, Dict, List, Any from mcp.server.fastmcp import Context async def list_entities( page: int = 1, per_page: int = 20, search: Optional[str] = None, ctx: Context = None ) -> Dict[str, Any]: """List entities with pagination.""" pass ``` ### Error Handling ```python from ..utils.errors import MCPValidationError, MCPAuthenticationError async def get_entity(entity_id: str, ctx: Context) -> dict: try: # Validate input entity_id = validate_uuid(entity_id, "entity_id") # Extract token token = await extract_user_token(ctx) # Make API call data = await api_client.get(f"/entities/{entity_id}", token=token) return {"success": True, "entity": data} except MCPValidationError as e: return {"success": False, "error": str(e)} except MCPAuthenticationError: return {"success": False, "error": "Not authenticated"} except Exception as e: logger.error(f"Unexpected error", error=str(e)) return {"success": False, "error": f"Unexpected error: {str(e)}"} ``` ### Logging ```python import structlog logger = structlog.get_logger(__name__) async def some_function(): log = logger.bind(action="some_function", param="value") log.info("Starting operation") try: # Do something log.info("Operation successful", result="success") except Exception as e: log.error("Operation failed", error=str(e)) ``` ## Adding New MCP Tools ### Step-by-Step Guide #### 1. Create Tool Function ```python # src/finizi_b4b_mcp/tools/new_tool.py from mcp.server.fastmcp import Context from ..auth.token_handler import extract_user_token from ..client.api_client import get_api_client async def new_tool_function( param1: str, param2: int = 10, ctx: Context = None ) -> dict: """ Tool description here. Args: param1: Description of param1 param2: Description of param2 ctx: MCP context (automatically provided) Returns: Dictionary with success status and data """ logger = structlog.get_logger(__name__) log = logger.bind(action="new_tool", param1=param1) try: # Extract token token = await extract_user_token(ctx) # Validate inputs # ... # Make API call api_client = get_api_client() data = await api_client.get("/endpoint", token=token) # Format response return { "success": True, "data": data } except Exception as e: log.error("Tool failed", error=str(e)) return { "success": False, "error": str(e) } ``` #### 2. Register Tool in Server ```python # src/finizi_b4b_mcp/server.py from .tools import new_tool # Register the tool mcp.tool()(new_tool.new_tool_function) logger.info("Registered new tool") ``` #### 3. Add Tests ```python # tests/test_new_tool.py import pytest from unittest.mock import AsyncMock, patch from finizi_b4b_mcp.tools.new_tool import new_tool_function @pytest.mark.asyncio async def test_new_tool_success(): """Test successful execution of new tool.""" # Mock context ctx = AsyncMock() ctx.session.metadata = {"user_token": "test_token"} # Mock API response with patch('finizi_b4b_mcp.client.api_client.ApiClient.get') as mock_get: mock_get.return_value = {"test": "data"} result = await new_tool_function("param", ctx=ctx) assert result["success"] is True assert result["data"]["test"] == "data" ``` #### 4. Update Documentation - Add to README.md tool list - Update API_MAPPING.md with endpoint mapping - Add usage examples ### Tool Design Checklist - [ ] Clear, descriptive function name - [ ] Comprehensive docstring - [ ] Input validation - [ ] Token extraction (if auth required) - [ ] Error handling for all failure modes - [ ] Structured logging - [ ] Consistent response format - [ ] Unit tests with mocks - [ ] Integration tests (optional) - [ ] Documentation updated ## Testing Guide ### Running Tests #### All Tests ```bash # Basic run pytest # With verbose output pytest -v # With coverage pytest --cov=src/finizi_b4b_mcp --cov-report=html # Specific test file pytest tests/test_auth.py # Specific test function pytest tests/test_auth.py::test_login_success # With print statements pytest -s ``` ### Writing Tests #### Test Structure ```python import pytest from unittest.mock import AsyncMock, patch, MagicMock class TestEntityTools: """Test suite for entity management tools.""" @pytest.fixture def mock_ctx(self): """Create mock MCP context.""" ctx = AsyncMock() ctx.session.metadata = { "user_token": "test_token", "user_id": "test_user_id" } return ctx @pytest.fixture def mock_api_client(self): """Create mock API client.""" with patch('finizi_b4b_mcp.client.api_client.get_api_client') as mock: client = AsyncMock() mock.return_value = client yield client @pytest.mark.asyncio async def test_list_entities(self, mock_ctx, mock_api_client): """Test listing entities.""" # Arrange mock_api_client.get.return_value = { "items": [{"id": "123", "name": "Test"}], "total": 1 } # Act from finizi_b4b_mcp.tools.entities import list_entities result = await list_entities(ctx=mock_ctx) # Assert assert result["success"] is True assert len(result["items"]) == 1 mock_api_client.get.assert_called_once() ``` #### Testing Patterns **Success Cases:** ```python async def test_operation_success(): """Test successful operation.""" # Test with valid inputs # Verify correct API calls # Check response format ``` **Error Cases:** ```python async def test_operation_auth_error(): """Test authentication error handling.""" ctx = AsyncMock() ctx.session.metadata = {} # No token result = await some_tool(ctx=ctx) assert result["success"] is False assert "authenticated" in result["error"].lower() ``` **Edge Cases:** ```python async def test_operation_empty_response(): """Test handling of empty responses.""" # Mock empty API response # Verify graceful handling ``` ### Code Coverage #### Generate Coverage Report ```bash # HTML report pytest --cov=src/finizi_b4b_mcp --cov-report=html open htmlcov/index.html # Terminal report pytest --cov=src/finizi_b4b_mcp --cov-report=term # XML report (for CI) pytest --cov=src/finizi_b4b_mcp --cov-report=xml ``` #### Coverage Goals - Minimum 80% overall coverage - 100% coverage for validators - 90% coverage for tools - Focus on critical paths ## Debugging Tips ### Enable Debug Logging ```python # In .env LOG_LEVEL=DEBUG # Or programmatically import structlog structlog.configure( processors=[ structlog.processors.add_log_level, structlog.processors.TimeStamper(fmt="iso"), structlog.dev.ConsoleRenderer(colors=True) ] ) ``` ### Interactive Debugging #### Using IPython ```python # Add breakpoint in code import IPython; IPython.embed() # Or use debugger import pdb; pdb.set_trace() ``` #### Using VSCode ```json // .vscode/launch.json { "version": "0.2.0", "configurations": [ { "name": "Debug MCP Server", "type": "python", "request": "launch", "program": "${workspaceFolder}/run_server.py", "console": "integratedTerminal", "env": { "B4B_API_BASE_URL": "http://localhost:8000", "LOG_LEVEL": "DEBUG" } } ] } ``` ### HTTP Request Debugging #### Enable httpx Logging ```python import httpx import logging logging.basicConfig(level=logging.DEBUG) httpx_logger = logging.getLogger("httpx") httpx_logger.setLevel(logging.DEBUG) ``` #### Use HTTP Proxy ```python # In api_client.py client = httpx.AsyncClient( proxies={"http://": "http://localhost:8888"}, verify=False # For proxy SSL ) ``` ### Common Debugging Commands ```bash # Check Python version python --version # List installed packages uv pip list # Check environment variables python -c "import os; print(os.environ.get('B4B_API_BASE_URL'))" # Test import python -c "from finizi_b4b_mcp import mcp; print(mcp)" # Run single tool test python -c " import asyncio from finizi_b4b_mcp.tools.auth import login asyncio.run(login('+84909495665', 'pass')) " ``` ## Common Issues and Solutions ### Issue: Import Error **Error:** ``` ModuleNotFoundError: No module named 'finizi_b4b_mcp' ``` **Solution:** ```bash # Install in editable mode uv pip install -e "." # Verify installation python -c "import finizi_b4b_mcp" ``` ### Issue: Connection Refused **Error:** ``` httpx.ConnectError: [Errno 111] Connection refused ``` **Solution:** 1. Check B4B API is running 2. Verify API URL in .env 3. Check firewall settings 4. Test with curl: ```bash curl http://localhost:8000/api/v1/health ``` ### Issue: Authentication Failed **Error:** ``` MCPAuthenticationError: Token expired or invalid ``` **Solution:** 1. Call login tool again 2. Check token expiration 3. Verify credentials 4. Clear session and retry ### Issue: Rate Limiting **Error:** ``` HTTPStatusError: 429 Too Many Requests ``` **Solution:** 1. Implement request throttling 2. Add delays between requests 3. Use exponential backoff 4. Contact API admin for limit increase ### Issue: Timeout Errors **Error:** ``` httpx.ReadTimeout: The read operation timed out ``` **Solution:** ```python # Increase timeout in .env API_TIMEOUT=60 API_CONNECT_TIMEOUT=20 # Or in code client = httpx.AsyncClient( timeout=httpx.Timeout(60.0, connect=20.0) ) ``` ### Issue: SSL Certificate Error **Error:** ``` httpx.ConnectError: [SSL: CERTIFICATE_VERIFY_FAILED] ``` **Solution:** ```python # Development only - disable SSL verification client = httpx.AsyncClient(verify=False) # Production - add certificate client = httpx.AsyncClient(verify="/path/to/cert.pem") ``` ## Performance Considerations ### Connection Pooling ```python # api_client.py configuration limits = httpx.Limits( max_keepalive_connections=20, max_connections=100, keepalive_expiry=30.0 ) ``` ### Async Optimization ```python # Parallel requests import asyncio async def fetch_multiple(): tasks = [ api_client.get("/endpoint1"), api_client.get("/endpoint2"), api_client.get("/endpoint3") ] results = await asyncio.gather(*tasks) return results ``` ### Caching Strategy ```python from functools import lru_cache from datetime import datetime, timedelta class CachedData: def __init__(self, data, expiry): self.data = data self.expiry = expiry def is_valid(self): return datetime.now() < self.expiry cache = {} async def get_with_cache(key: str, fetch_func): if key in cache and cache[key].is_valid(): return cache[key].data data = await fetch_func() cache[key] = CachedData( data, datetime.now() + timedelta(minutes=5) ) return data ``` ### Pagination Best Practices ```python async def fetch_all_pages(endpoint: str, token: str): """Fetch all pages efficiently.""" all_items = [] page = 1 per_page = 100 # Max allowed while True: data = await api_client.get( endpoint, params={"page": page, "per_page": per_page}, token=token ) all_items.extend(data["items"]) if page >= data["pages"]: break page += 1 return all_items ``` ### Memory Management ```python # Stream large responses async def stream_large_data(): async with httpx.AsyncClient() as client: async with client.stream("GET", url) as response: async for chunk in response.aiter_bytes(): process_chunk(chunk) ``` ## Security Best Practices ### Token Security ```python # Never log tokens logger.info("User authenticated", user_id=user_id) # Good logger.info(f"Token: {token}") # Bad! # Clear tokens on logout ctx.session.metadata.clear() # Use secure token storage import secrets token_key = secrets.token_urlsafe(32) ``` ### Input Validation ```python import re from uuid import UUID def validate_phone(phone: str) -> str: """Validate phone number format.""" pattern = r'^\+\d{10,15}$' if not re.match(pattern, phone): raise ValueError("Invalid phone format") return phone def validate_uuid(value: str) -> str: """Validate UUID format.""" try: UUID(value, version=4) return value except ValueError: raise ValueError(f"Invalid UUID: {value}") ``` ### SQL Injection Prevention ```python # Use parameterized queries (if using DB) # Never concatenate user input into queries # Bad query = f"SELECT * FROM users WHERE id = '{user_id}'" # Good query = "SELECT * FROM users WHERE id = %s" cursor.execute(query, (user_id,)) ``` ### XSS Prevention ```python import html def sanitize_output(text: str) -> str: """Escape HTML special characters.""" return html.escape(text) ``` ### Rate Limiting ```python from collections import defaultdict from datetime import datetime, timedelta class RateLimiter: def __init__(self, max_requests=100, window_seconds=60): self.max_requests = max_requests self.window = timedelta(seconds=window_seconds) self.requests = defaultdict(list) def allow_request(self, user_id: str) -> bool: now = datetime.now() cutoff = now - self.window # Clean old requests self.requests[user_id] = [ t for t in self.requests[user_id] if t > cutoff ] # Check limit if len(self.requests[user_id]) >= self.max_requests: return False self.requests[user_id].append(now) return True ``` ### Environment Security ```bash # .env should never be committed echo ".env" >> .gitignore # Use different .env files for different environments .env.development .env.staging .env.production # Rotate secrets regularly # Use secret management services in production ``` ## Development Workflow ### Git Workflow ```bash # Create feature branch git checkout -b feature/new-tool # Make changes vim src/finizi_b4b_mcp/tools/new_tool.py # Run tests pytest tests/ # Format code ruff format . # Lint code ruff check --fix . # Commit changes git add . git commit -m "feat: Add new tool for X functionality" # Push branch git push origin feature/new-tool # Create pull request gh pr create ``` ### Pre-commit Hooks ```bash # Install pre-commit pip install pre-commit # Create .pre-commit-config.yaml cat > .pre-commit-config.yaml << EOF repos: - repo: https://github.com/charliermarsh/ruff-pre-commit rev: v0.1.0 hooks: - id: ruff args: [--fix] - id: ruff-format - repo: https://github.com/pre-commit/pre-commit-hooks rev: v4.5.0 hooks: - id: trailing-whitespace - id: end-of-file-fixer - id: check-yaml - id: check-json - id: check-added-large-files EOF # Install hooks pre-commit install # Run manually pre-commit run --all-files ``` ### Continuous Integration ```yaml # .github/workflows/ci.yml name: CI on: [push, pull_request] jobs: test: runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 - name: Set up Python uses: actions/setup-python@v4 with: python-version: '3.11' - name: Install UV run: curl -LsSf https://astral.sh/uv/install.sh | sh - name: Install dependencies run: | uv venv source .venv/bin/activate uv pip install -e ".[dev]" - name: Run tests run: | source .venv/bin/activate pytest --cov=src/finizi_b4b_mcp - name: Lint run: | source .venv/bin/activate ruff check . ``` --- **Last Updated**: October 2024 **Version**: 1.0.0 **Maintained by**: Finizi Development Team