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:
Note: You may need to restart your Python kernel after installation:
# For Jupyter/Databricks
dbutils.library.restartPython()
Quick Start
1. Import Required Libraries
import os
from mcp import ClientSession
from mcp.client.sse import sse_client
import asyncio
import json
2. Basic Connection Example
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())
See all 27 lines
Common Use Cases
Matching Businesses
Find business IDs for companies by name or domain:
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())
See all 33 lines
Finding Prospects at Companies
Search for employees with specific roles:
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())
See all 48 lines
Get email addresses and phone numbers for prospects:
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())
See all 31 lines
Advanced Patterns
Using Autocomplete for Valid Filter Values
Many filters require valid values from the autocomplete tool:
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())
See all 44 lines
Comprehensive Company Analysis
Combine multiple enrichment tools for detailed insights:
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())
See all 43 lines
Error Handling
Implement robust error handling for production use:
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 } " )
See all 39 lines
Best Practices
1. Environment Variables
Always use environment variables for API keys:
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:
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:
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:
# 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
}
)
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:
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
Support
If you encounter any issues or have questions:
Note: Remember to replace'your-api-key-here' with your actual AgentSource API key.