Python SDK Implementation Guide

Overview

This guide demonstrates how to use AgentSource MCP with the Python MCP SDK. The SDK provides a native Python interface for connecting to MCP servers, making it ideal for building custom applications, data pipelines, and automation tools.

Prerequisites

Python 3.8 or higher
AgentSource API key
Basic understanding of asynchronous Python programming

Installation

Install the MCP Python SDK:

Bash

pip install mcp

Note: You may need to restart your Python kernel after installation:

Python

# For Jupyter/Databricks
dbutils.library.restartPython()

Quick Start

1. Import Required Libraries

Python

import os
from mcp import ClientSession
from mcp.client.sse import sse_client
import asyncio
import json

2. Basic Connection Example

Python

async def connect_to_agentsource():
    # Configure your API key
    api_key = os.environ.get('EXPLORIUM_API_KEY', 'your-api-key-here')
    headers = {
        "Authorization": f"Bearer {api_key}",
    }

    # Connect to the MCP endpoint
    async with sse_client(
        url="https://mcp.explorium.ai/sse",
        headers=headers
    ) as streams:
        read_stream, write_stream = streams

        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            print("✓ Successfully connected to AgentSource MCP!")

            # List available tools
            tools = await session.list_tools()
            print(f"Available tools: {len(tools.tools)}")

            for tool in tools.tools[:5]:  # Show first 5 tools
                print(f"  • {tool.name}: {tool.description[:60]}...")

# Run the async function
asyncio.run(connect_to_agentsource())

Common Use Cases

Matching Businesses

Find business IDs for companies by name or domain:

Python

async def match_businesses_example():
    api_key = os.environ.get('EXPLORIUM_API_KEY', 'your-api-key')
    headers = {"Authorization": f"Bearer {api_key}"}

    async with sse_client(
        url="https://mcp.explorium.ai/sse",
        headers=headers
    ) as streams:
        read_stream, write_stream = streams

        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()

            # Prepare businesses to match
            match_input = {
                "businesses_to_match": [
                    {"name": "Google"},
                    {"domain": "microsoft.com"},
                    {"name": "Amazon", "domain": "amazon.com"}
                ]
            }

            # Call the match_businesses tool
            result = await session.call_tool("match_businesses", arguments=match_input)

            # Process results
            if hasattr(result, 'content') and result.content:
                data = json.loads(result.content[0].text)
                print(f"Matched {data['total_matches']} businesses:")
                for business in data['matched_businesses']:
                    print(f"  • {business['input']} <Icon icon="arrow-right" /> ID: {business['business_id']}")

asyncio.run(match_businesses_example())

Finding Prospects at Companies

Search for employees with specific roles:

Python

async def find_prospects_example():
    api_key = os.environ.get('EXPLORIUM_API_KEY', 'your-api-key')
    headers = {"Authorization": f"Bearer {api_key}"}

    async with sse_client(
        url="https://mcp.explorium.ai/sse",
        headers=headers
    ) as streams:
        read_stream, write_stream = streams

        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()

            # First, match the company
            match_result = await session.call_tool(
                "match_businesses",
                arguments={"businesses_to_match": [{"name": "Salesforce"}]}
            )

            # Extract business ID from match result
            match_data = json.loads(match_result.content[0].text)
            business_id = match_data['matched_businesses'][0]['business_id']

            # Search for VPs in Engineering
            prospects_input = {
                "filters": {
                    "business_id": [business_id],
                    "job_level": ["vp"],
                    "job_department": ["Engineering"]
                },
                "size": 20,
                "page_size": 10,
                "page": 1
            }

            # Fetch prospects
            prospects_result = await session.call_tool(
                "fetch_prospects",
                arguments=prospects_input
            )

            # Process and display results
            prospects_data = json.loads(prospects_result.content[0].text)
            print(f"Found {len(prospects_data['data'])} VPs in Engineering:")
            for prospect in prospects_data['data'][:5]:
                print(f"  • {prospect['full_name']} - {prospect['job_title']}")

asyncio.run(find_prospects_example())

Enriching Contact Information

Get email addresses and phone numbers for prospects:

Python

async def enrich_contacts_example():
    api_key = os.environ.get('EXPLORIUM_API_KEY', 'your-api-key')
    headers = {"Authorization": f"Bearer {api_key}"}

    async with sse_client(
        url="https://mcp.explorium.ai/sse",
        headers=headers
    ) as streams:
        read_stream, write_stream = streams

        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()

            # Assume we have prospect IDs from a previous search
            prospect_ids = ["prospect_id_1", "prospect_id_2"]

            # Enrich contact information
            enrich_result = await session.call_tool(
                "enrich_prospects_contacts_information",
                arguments={"prospect_ids": prospect_ids}
            )

            # Display enriched data
            enrich_data = json.loads(enrich_result.content[0].text)
            for prospect in enrich_data['data']:
                if 'emails' in prospect['data']:
                    print(f"Contact: {prospect['prospect_id']}")
                    for email in prospect['data']['emails']:
                        print(f"  • Email ({email['type']}): {email['email']}")

asyncio.run(enrich_contacts_example())

Advanced Patterns

Using Autocomplete for Valid Filter Values

Many filters require valid values from the autocomplete tool:

Python

async def search_with_autocomplete():
    api_key = os.environ.get('AGENTSOURCE_API_KEY', 'your-api-key')
    headers = {"Authorization": f"Bearer {api_key}"}

    async with sse_client(
        url="https://mcp.explorium.ai/sse",
        headers=headers
    ) as streams:
        read_stream, write_stream = streams

        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()

            # Get valid LinkedIn® categories for "software"
            autocomplete_result = await session.call_tool(
                "autocomplete",
                arguments={
                    "field": "linkedin_category",
                    "query": "software"
                }
            )

            # Parse autocomplete results
            auto_data = json.loads(autocomplete_result.content[0].text)
            categories = [item['value'] for item in auto_data['data']]

            # Use the validated categories in search
            search_result = await session.call_tool(
                "fetch_businesses",
                arguments={
                    "filters": {
                        "linkedin_category": categories[:2],  # Use first 2 categories
                        "company_size": ["51-200", "201-500"],
                        "country_code": ["US"]
                    },
                    "size": 10
                }
            )

            # Process results
            search_data = json.loads(search_result.content[0].text)
            print(f"Found {search_data['total_results']} companies")

asyncio.run(search_with_autocomplete())

Comprehensive Company Analysis

Combine multiple enrichment tools for detailed insights:

Python

async def comprehensive_company_analysis():
    api_key = os.environ.get('AGENTSOURCE_API_KEY', 'your-api-key')
    headers = {"Authorization": f"Bearer {api_key}"}

    async with sse_client(
        url="https://mcp.explorium.ai/sse",
        headers=headers
    ) as streams:
        read_stream, write_stream = streams

        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()

            # Match a company
            match_result = await session.call_tool(
                "match_businesses",
                arguments={"businesses_to_match": [{"domain": "tesla.com"}]}
            )

            match_data = json.loads(match_result.content[0].text)
            business_id = match_data['matched_businesses'][0]['business_id']

            # Gather multiple types of enrichment data
            enrichment_tools = [
                ("enrich_businesses_firmographics", "Company Overview"),
                ("enrich_businesses_technographics", "Technology Stack"),
                ("enrich_businesses_workforce_trends", "Workforce Trends"),
                ("enrich_businesses_funding_and_acquisitions", "Funding History")
            ]

            for tool_name, description in enrichment_tools:
                print(f"\n{description}:")
                result = await session.call_tool(
                    tool_name,
                    arguments={"business_ids": [business_id]}
                )

                # Process and display key insights
                data = json.loads(result.content[0].text)
                # Display relevant information based on the tool
                print(f"  ✓ Data retrieved successfully")

asyncio.run(comprehensive_company_analysis())

Error Handling

Implement robust error handling for production use:

Python

async def safe_mcp_call():
    api_key = os.environ.get('AGENTSOURCE_API_KEY')
    if not api_key:
        raise ValueError("AGENTSOURCE_API_KEY environment variable not set")

    headers = {"Authorization": f"Bearer {api_key}"}

    try:
        async with sse_client(
            url="https://mcp.explorium.ai/sse",
            headers=headers
        ) as streams:
            read_stream, write_stream = streams

            async with ClientSession(read_stream, write_stream) as session:
                try:
                    await session.initialize()

                    # Your MCP operations here
                    result = await session.call_tool(
                        "match_businesses",
                        arguments={"businesses_to_match": [{"name": "Example Corp"}]}
                    )

                    return result

                except Exception as e:
                    print(f"Error during tool execution: {e}")
                    raise

    except Exception as e:
        print(f"Connection error: {e}")
        raise

# Run with error handling
try:
    result = asyncio.run(safe_mcp_call())
except Exception as e:
    print(f"Failed to execute MCP call: {e}")

Best Practices

1. Environment Variables

Always use environment variables for API keys:

Python

import os
api_key = os.environ.get('EXPLORIUM_API_KEY')
if not api_key:
    raise ValueError("Please set EXPLORIUM_API_KEY environment variable")

2. Connection Management

Use context managers to ensure proper connection cleanup:

Python

async with sse_client(url, headers) as streams:
    async with ClientSession(*streams) as session:
        # Your operations here
        pass  # Connection automatically closed

3. Result Parsing

Always validate and parse results safely:

Python

def parse_tool_result(result):
    """Safely parse tool results"""
    if hasattr(result, 'content') and result.content:
        try:
            return json.loads(result.content[0].text)
        except (json.JSONDecodeError, IndexError, AttributeError):
            print("Failed to parse result")
            return None
    return None

4. Batch Operations

When processing multiple items, use appropriate page sizes:

Python

# Good: Process in reasonable batches
page_size = 50
for page in range(1, total_pages + 1):
    result = await session.call_tool(
        "fetch_businesses",
        arguments={
            "filters": {...},
            "page_size": page_size,
            "page": page
        }
    )

Available Tools Reference

The MCP SDK provides access to all AgentSource tools:

Business Tools: match_businesses, fetch_businesses, enrich_businesses_*
Prospect Tools: match_prospects, fetch_prospects, enrich_prospects_*
Utility Tools: autocomplete, fetch_businesses_events, fetch_prospects_events

Use session.list_tools() to discover all available tools and their parameters.

Debugging Tips

Enable verbose logging to troubleshoot issues:

Python

import logging

# Set up logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

# Log tool calls
logger.debug(f"Calling tool: {tool_name}")
logger.debug(f"Arguments: {json.dumps(arguments, indent=2)}")

Next Steps

Explore the complete AgentSource MCP documentation
Check out the OpenAI implementation guide for AI agent integration
Learn about advanced filtering and search capabilities
Build production-ready data pipelines with MCP

Support

If you encounter any issues or have questions:

Check the AgentSource MCP reference
Contact our support team at support@explorium.ai

Note:Remember to replace'your-api-key-here' with your actual AgentSource API key.

​Overview

​Prerequisites

​Installation

​Quick Start

​1. Import Required Libraries

​2. Basic Connection Example

​Common Use Cases

​Matching Businesses

​Finding Prospects at Companies

​Enriching Contact Information

​Advanced Patterns

​Using Autocomplete for Valid Filter Values

​Comprehensive Company Analysis

​Error Handling

​Best Practices

​1. Environment Variables

​2. Connection Management

​3. Result Parsing

​4. Batch Operations

​Available Tools Reference

​Debugging Tips

​Next Steps

​Support

Overview

Prerequisites

Installation

Quick Start

1. Import Required Libraries

2. Basic Connection Example

Common Use Cases

Matching Businesses

Finding Prospects at Companies

Enriching Contact Information

Advanced Patterns

Using Autocomplete for Valid Filter Values

Comprehensive Company Analysis

Error Handling

Best Practices

1. Environment Variables

2. Connection Management

3. Result Parsing

4. Batch Operations

Available Tools Reference

Debugging Tips

Next Steps

Support