OpenAPI Lambda MCP Server

#!/usr/bin/env python3 """Pytest tests for Zoho CRM MCP Server. Tests: 1. List available tools 2. Search for a contact by email Run with: pytest packages/apis/zoho-crm/test_zoho_mcp.py -v OAuth credentials are loaded from .env file or environment variables. """ import json import os from pathlib import Path from urllib.request import Request, urlopen from urllib.error import HTTPError from urllib.parse import urlencode import pytest # Try to load .env file if it exists try: from dotenv import load_dotenv env_file = Path(__file__).parent / '.env' if env_file.exists(): load_dotenv(env_file) print(f"✅ Loaded credentials from {env_file}") except ImportError: pass # python-dotenv not installed, will use environment variables @pytest.fixture(scope="module") def mcp_url(): """Get MCP server URL from environment and add OAuth credentials as query params.""" url = os.getenv("MCP_SERVER_URL") if not url: pytest.skip("MCP_SERVER_URL environment variable not set") # Add OAuth credentials if provided client_id = os.getenv("ZOHO_CLIENT_ID") client_secret = os.getenv("ZOHO_CLIENT_SECRET") refresh_token = os.getenv("ZOHO_REFRESH_TOKEN") if client_id and client_secret and refresh_token: # Add credentials as query parameters params = { 'client_id': client_id, 'client_secret': client_secret, 'refresh_token': refresh_token } separator = "&" if "?" in url else "?" url = f"{url}{separator}{urlencode(params)}" print(f"🔐 Added OAuth credentials to URL") else: print(f"⚠️ OAuth credentials not found - some tests will be skipped") return url @pytest.fixture(scope="module") def test_email(): """Get test email from environment.""" return os.getenv("TEST_EMAIL", "guillaume@ingeno.ca") def make_mcp_request(url: str, method: str, params: dict = None) -> dict: """Make an MCP request to the server.""" payload = { "jsonrpc": "2.0", "method": method, "id": 1, } if params: payload["params"] = params request = Request( url, data=json.dumps(payload).encode('utf-8'), headers={ 'Content-Type': 'application/json', 'Accept': 'application/json, text/event-stream' }, method='POST' ) try: with urlopen(request) as response: content = response.read().decode('utf-8') # Handle SSE (Server-Sent Events) format if content.startswith('event:'): # Parse SSE format: "event: message\r\ndata: {json}\r\n\r\n" for line in content.split('\n'): if line.startswith('data: '): json_data = line[6:] # Remove "data: " prefix return json.loads(json_data) # Handle plain JSON return json.loads(content) except HTTPError as e: error_body = e.read().decode('utf-8') pytest.fail(f"HTTP Error {e.code}: {error_body}") def test_list_tools(mcp_url): """Test listing available tools.""" response = make_mcp_request(mcp_url, "tools/list") assert "result" in response, f"Expected 'result' in response, got: {response}" assert "tools" in response["result"], "Expected 'tools' in result" tools = response["result"]["tools"] assert len(tools) > 0, "Expected at least one tool" # Print tools for debugging print(f"\n✅ Found {len(tools)} tools:") for tool in tools[:5]: print(f"\n Tool: {tool.get('name')}") print(f" Description: {tool.get('description', 'N/A')}") if len(tools) > 5: print(f"\n ... and {len(tools) - 5} more tools") def test_search_contact(mcp_url, test_email): """Test searching for a contact by email.""" # First, list tools to find the Search_Records tool tools_response = make_mcp_request(mcp_url, "tools/list") tools = tools_response.get("result", {}).get("tools", []) # Find the Search_Records tool search_tool = None for tool in tools: if tool.get("name") == "Search_Records": search_tool = tool break assert search_tool is not None, "Could not find Search_Records tool" print(f"\n📋 Using tool: {search_tool['name']}") # Call the tool with Contacts module response = make_mcp_request( mcp_url, "tools/call", { "name": search_tool["name"], "arguments": { "module_api_name": "Contacts", "email": test_email } } ) # Check for errors assert "error" not in response, f"Got error: {response.get('error')}" assert "result" in response, f"Expected 'result' in response, got: {response}" result = response["result"] # Check if the response indicates an error (isError flag from MCP) if result.get("isError"): # Extract error message from content content = result.get("content", []) if content and content[0].get("type") == "text": error_text = content[0].get("text", "") print(f"\n❌ Tool execution error: {error_text}") if "AUTHENTICATION_FAILURE" in error_text or "Authentication failed" in error_text: pytest.skip(f"Authentication failed - check OAuth credentials: {error_text}") pytest.fail(f"Tool execution failed: {error_text}") # Verify we got content assert "content" in result, "Expected 'content' in result" content = result["content"] assert len(content) > 0, "Expected at least one content item" assert content[0].get("type") == "text", "Expected text content type" # Parse the response text (which should be JSON) response_text = content[0].get("text", "") assert response_text, "Expected non-empty response text" response_data = json.loads(response_text) # Verify we got data assert "result" in response_data, f"Expected 'result' in response data: {response_data}" assert "data" in response_data["result"], f"Expected 'data' in result: {response_data['result']}" contacts = response_data["result"]["data"] assert isinstance(contacts, list), f"Expected 'data' to be a list, got {type(contacts)}" assert len(contacts) > 0, f"Expected at least one contact for email {test_email}, got 0" # Verify the contact has the expected email first_contact = contacts[0] contact_email = first_contact.get("Email", "").lower() assert contact_email == test_email.lower(), \ f"Expected contact email to be {test_email}, got {contact_email}" print(f"\n✅ Search successful! Found {len(contacts)} contact(s)") print(f"Contact details:") print(f" Email: {first_contact.get('Email')}") print(f" Name: {first_contact.get('Full_Name', 'N/A')}") print(f" ID: {first_contact.get('id', 'N/A')}")