Python SDK Implementation Guide

Overview

The Explorium MCP (Model Context Protocol) server provides a standardized way to integrate Explorium's data enrichment capabilities with AI applications. This guide demonstrates how to connect to the Explorium MCP server using the Python SDK.

Prerequisites

Before you begin, ensure you have:

Python 3.11 or higher
An Explorium API key
Access to a Python environment (local, Databricks, or Jupyter)

Installation

Install the required MCP package:

pip install mcp

For Databricks environments:

%pip install mcp
dbutils.library.restartPython()

Basic Implementation

1. Import Required Libraries

import os
import json
import asyncio
from typing import Any
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

2. Connect to the MCP Server

The Explorium MCP server uses the Streamable HTTP transport protocol. Here's how to establish a connection:

async def connect_to_explorium_mcp():
    # Configure your API key
    api_key = os.environ.get('EXPLORIUM_API_KEY', 'your-api-key-here')
    
    # Set up headers with API key
    headers = {
        "api_key": api_key
    }
    
    # Connect to the MCP endpoint
    async with streamablehttp_client(
        url="https://mcp.explorium.ai/mcp",
        headers=headers
    ) as (read_stream, write_stream, _):  # Note: returns 3 values
        
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            print("Successfully connected to Explorium MCP!")
            
            # Your code here
            return session

Important Note: The streamablehttp_client returns a tuple of three values: (read_stream, write_stream, _). The third value is typically unused but must be unpacked to avoid errors.

3. List Available Tools

Once connected, you can discover the available Explorium tools:

async def list_explorium_tools():
    api_key = os.environ.get('EXPLORIUM_API_KEY', 'your-api-key-here')
    headers = {"api_key": api_key}
    
    async with streamablehttp_client(
        url="https://mcp.explorium.ai/mcp",
        headers=headers
    ) as (read_stream, write_stream, _):
        
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            
            # List all available tools
            tools = await session.list_tools()
            
            print(f"Found {len(tools.tools)} tools:")
            for tool in tools.tools:
                print(f"  - {tool.name}: {tool.description}")
            
            return tools.tools

4. Call MCP Tools

Here's an example of calling the match-business tool:

async def match_businesses():
    api_key = os.environ.get('EXPLORIUM_API_KEY', 'your-api-key-here')
    headers = {"api_key": api_key}
    
    async with streamablehttp_client(
        url="https://mcp.explorium.ai/mcp",
        headers=headers
    ) as (read_stream, write_stream, _):
        
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            
            # Prepare the input for business matching
            match_input = {
                "businesses_to_match": [
                    {"name": "Google"},
                    {"domain": "microsoft.com"},
                    {"name": "Amazon", "domain": "amazon.com"}
                ]
            }
            
            # Call the match-business tool
            result = await session.call_tool(
                "match-business", 
                arguments=match_input
            )
            
            # Handle the response
            if hasattr(result, 'content') and result.content:
                content = result.content[0]
                if hasattr(content, 'text'):
                    # Parse JSON response
                    data = json.loads(content.text)
                    return data
            
            return None

Complete Working Example

Here's a full example that connects to the MCP server, lists tools, and performs a business match:

import os
import json
import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async def run_explorium_mcp_example():
    """
    Complete example demonstrating Explorium MCP usage
    """
    # Get API key from environment or use default
    api_key = os.environ.get('EXPLORIUM_API_KEY', 'your-api-key-here')
    headers = {"api_key": api_key}
    
    print("Connecting to Explorium MCP endpoint...")
    
    async with streamablehttp_client(
        url="https://mcp.explorium.ai/mcp",
        headers=headers
    ) as (read_stream, write_stream, _):
        
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            print("✓ Successfully connected and initialized!")
            
            # List available tools
            print("\nAvailable Tools:")
            tools = await session.list_tools()
            
            for tool in tools.tools[:10]:  # Show first 10 tools
                print(f"  - {tool.name}: {tool.description}")
            
            # Example: Match businesses
            print("\nPerforming business matching...")
            
            match_input = {
                "businesses_to_match": [
                    {"name": "Google"},
                    {"domain": "microsoft.com"},
                    {"name": "Amazon", "domain": "amazon.com"}
                ]
            }
            
            try:
                matched = await session.call_tool(
                    "match-business", 
                    arguments=match_input
                )
                
                # Process the response
                if hasattr(matched, 'content') and matched.content:
                    content = matched.content[0]
                    
                    if hasattr(content, 'text'):
                        data = json.loads(content.text)
                        print("\nMatch Results:")
                        print(json.dumps(data, indent=2))
                    else:
                        print(content)
                
                print("\n✓ Business matching completed successfully!")
                
            except Exception as e:
                print(f"\n✗ Error matching businesses: {e}")

# Run the example
if __name__ == "__main__":
    asyncio.run(run_explorium_mcp_example())

Available Tools

The Explorium MCP server typically provides the following tools:

match-business - Match and identify businesses
fetch-businesses - Retrieve business information
fetch-businesses-statistics - Get business statistics
fetch-businesses-events - Fetch business events
enrich-business - Enrich business data
match-prospects - Match prospect contacts
fetch-prospects - Retrieve prospect information
fetch-prospects-events - Fetch prospect events
fetch-prospects-statistics - Get prospect statistics
enrich-prospects - Enrich prospect data

Error Handling

Always implement proper error handling when working with the MCP client:

try:
    async with streamablehttp_client(
        url="https://mcp.explorium.ai/mcp",
        headers=headers
    ) as (read_stream, write_stream, _):
        # Your code here
        pass
        
except ValueError as e:
    print(f"Connection error: {e}")
    # Common issue: not unpacking all three values from streamablehttp_client
    
except Exception as e:
    print(f"Unexpected error: {e}")

Common Issues and Solutions

Issue 1: ValueError - too many values to unpack

Error: ValueError: too many values to unpack (expected 2)

Solution: The streamablehttp_client returns three values, not two. Always unpack as:

async with streamablehttp_client(...) as (read_stream, write_stream, _):

Issue 2: Authentication Error

Error: Authentication failed

Solution: Verify your API key is correct and properly formatted in the headers:

headers = {"api_key": "your-actual-api-key"}

Environment Variables

For production use, always store your API key as an environment variable:

export EXPLORIUM_API_KEY="your-api-key-here"

Or in Databricks:

# Using Databricks secrets
api_key = dbutils.secrets.get(scope="explorium", key="api-key")

Support

For additional support or questions about the Explorium MCP integration, please contact your Explorium representative or visit the Explorium Developer Portal.

API Endpoint

Production Endpoint: https://mcp.explorium.ai/mcp
Transport Protocol: Streamable HTTP
Authentication: API key in request headers