LangGraph Integration with Explorium MCP

pip install langgraph langchain_anthropic langchain_mcp_adapters

%pip install langgraph
%pip install langchain_anthropic
%pip install langchain_mcp_adapters

dbutils.library.restartPython()

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import load_mcp_tools

async def quick_start():
    # Initialize MCP client
    client = MultiServerMCPClient({
        "explorium": {
            "transport": "streamable_http",
            "url": "https://mcp.explorium.ai/mcp",
            "headers": {
                "api_key": "your-explorium-api-key"
            }
        }
    })
    
    # Create session and load tools
    async with client.session("explorium") as session:
        tools = await load_mcp_tools(session)
        print(f"Loaded {len(tools)} Explorium tools")
        for tool in tools[:5]:
            print(f"  - {tool.name}")

# Run the example
asyncio.run(quick_start())

import os
import asyncio
from typing import Annotated, TypedDict, cast

# LangGraph imports
from langgraph.func import END, START
from langgraph.graph import StateGraph
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.graph.message import add_messages

# Pydantic imports
from pydantic import BaseModel

# LangChain imports
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage
from langchain_anthropic import ChatAnthropic
from langchain_core.runnables import RunnableConfig

# MCP imports
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import load_mcp_tools

class ConfigSchema(TypedDict):
    """Configuration schema for API keys"""
    explorium_api_key: str
    anthropic_api_key: str


class AgentState(BaseModel):
    """State for the LangGraph agent"""
    messages: Annotated[list, add_messages]
    is_last_step: bool = False

EXPLORIUM_SYSTEM_PROMPT = """You are an AI assistant specialized in business intelligence and research.
You have access to Explorium data tools that can help find information about businesses and prospects.

When the user asks a question, use the appropriate tools to gather information and provide a helpful response.
Always be concise and focus on relevant business information.
"""

async def run_explorium_agent(config: RunnableConfig):
    """
    Runs the Explorium agent with active MCP session
    
    Args:
        config: Configuration containing API keys
    """
    # Get API keys from config
    explorium_api_key = config.get("configurable", {}).get("explorium_api_key")
    anthropic_api_key = config.get("configurable", {}).get("anthropic_api_key")
    
    if not explorium_api_key:
        raise ValueError("explorium_api_key is required in config")
    if not anthropic_api_key:
        raise ValueError("anthropic_api_key is required in config")
    
    print(f"Using Explorium API key: {explorium_api_key[:10]}...")
    
    # Initialize MCP client with Explorium configuration
    client = MultiServerMCPClient(
        {
            "explorium": {
                "transport": "streamable_http",
                "url": "https://mcp.explorium.ai/mcp",
                "headers": {
                    "api_key": explorium_api_key
                }
            }
        }
    )
    
    # Keep session active throughout the agent execution
    async with client.session("explorium") as session:
        # Load tools within the active session
        tools = await load_mcp_tools(session)
        print(f"Loaded {len(tools)} tools")
        
        # Print available tools for debugging
        print("Available tools:")
        for tool in tools[:10]:
            print(f"  - {tool.name}")
        
        # Initialize the LLM with Anthropic
        model = ChatAnthropic(
            model="claude-3-5-sonnet-20241022",  # Use a valid model name
            temperature=0.05,
            api_key=anthropic_api_key
        )
        
        # Define the reasoning node function
        async def reasoning_node(state: AgentState):
            bound_model = model.bind_tools(tools)
            system_prompt = SystemMessage(content=EXPLORIUM_SYSTEM_PROMPT)
            
            response = cast(
                AIMessage,
                await bound_model.ainvoke([system_prompt, *state.messages]),
            )
            
            # Handle the case when it's the last step and the model still wants to use a tool
            if state.is_last_step and response.tool_calls:
                return {
                    "messages": [
                        AIMessage(
                            id=response.id,
                            content="Sorry, I could not find an answer to your question in the specified number of steps.",
                        )
                    ]
                }
            
            # Return the model's response to be added to existing messages
            return {"messages": [response]}
        
        # Build the graph
        graph_builder = StateGraph(AgentState)
        
        # Add nodes
        graph_builder.add_node("reasoning_node", reasoning_node)
        graph_builder.add_node("tools", ToolNode(tools))
        
        # Add edges
        graph_builder.add_edge(START, "reasoning_node")
        graph_builder.add_conditional_edges(
            "reasoning_node",
            tools_condition,
            {"tools": "tools", END: END}
        )
        graph_builder.add_edge("tools", "reasoning_node")
        
        # Compile the graph
        graph = graph_builder.compile()
        
        # Initialize the graph with a user query
        user_query = "Find 10 product managers from Microsoft"
        
        # Create the initial state
        initial_state = AgentState(messages=[HumanMessage(content=user_query)])
        
        # Stream the execution for real-time updates
        async for event in graph.astream(initial_state, stream_mode="values"):
            if "messages" in event:
                latest_message = event["messages"][-1]
                
                if isinstance(latest_message, AIMessage):
                    if latest_message.content:
                        print(f"\nAI: {latest_message.content}")
                elif isinstance(latest_message, ToolMessage):
                    print(f"\nTool ({latest_message.name}): {latest_message.content[:200]}...")

async def run_langgraph():
    """Example usage of the Explorium LangGraph"""
    
    # Configuration with API keys
    config = {
        "configurable": {
            "explorium_api_key": "your-explorium-api-key",
            "anthropic_api_key": "your-anthropic-api-key"
        }
    }
    
    await run_explorium_agent(config)

# Run the agent
if __name__ == "__main__":
    asyncio.run(run_langgraph())

import os

config = {
    "configurable": {
        "explorium_api_key": os.environ.get("EXPLORIUM_API_KEY"),
        "anthropic_api_key": os.environ.get("ANTHROPIC_API_KEY")
    }
}

export EXPLORIUM_API_KEY="your-explorium-api-key"
export ANTHROPIC_API_KEY="your-anthropic-api-key"

# First, create secrets in Databricks CLI:
# databricks secrets create-scope --scope explorium
# databricks secrets put --scope explorium --key api-key
# databricks secrets create-scope --scope anthropic
# databricks secrets put --scope anthropic --key api-key

config = {
    "configurable": {
        "explorium_api_key": dbutils.secrets.get(scope="explorium", key="api-key"),
        "anthropic_api_key": dbutils.secrets.get(scope="anthropic", key="api-key")
    }
}

from langchain_google_vertexai import ChatVertexAI

model = ChatVertexAI(
    model_name="gemini-pro",
    project="your-gcp-project",
    location="us-central1"
)

async with client.session("explorium") as session:
    # Load all tools
    all_tools = await load_mcp_tools(session)
    
    # Filter to specific tools
    selected_tools = [
        tool for tool in all_tools 
        if tool.name in ["match-business", "fetch-prospects", "enrich-business"]
    ]
    
    # Use selected tools in your agent
    model = ChatAnthropic(
        model="claude-3-5-sonnet-20241022",
        api_key=anthropic_api_key
    )
    bound_model = model.bind_tools(selected_tools)

async for event in graph.astream(initial_state, stream_mode="values"):
    if "messages" in event:
        latest_message = event["messages"][-1]
        
        if isinstance(latest_message, AIMessage):
            # Handle AI responses
            if latest_message.content:
                print(f"AI Response: {latest_message.content}")
            if latest_message.tool_calls:
                print(f"Tools called: {[tc['name'] for tc in latest_message.tool_calls]}")
                
        elif isinstance(latest_message, ToolMessage):
            # Handle tool responses
            print(f"Tool '{latest_message.name}' returned data")
            # Process tool results here
            
        elif isinstance(latest_message, HumanMessage):
            # Handle user messages
            print(f"User said: {latest_message.content}")

async def run_with_error_handling():
    try:
        client = MultiServerMCPClient({
            "explorium": {
                "transport": "streamable_http",
                "url": "https://mcp.explorium.ai/mcp",
                "headers": {"api_key": "your-api-key"}
            }
        })
        
        async with client.session("explorium") as session:
            tools = await load_mcp_tools(session)
            # Your agent logic here
            
    except ConnectionError as e:
        print(f"Failed to connect to MCP server: {e}")
        print("Check your network connection and API endpoint")
        
    except ValueError as e:
        print(f"Configuration error: {e}")
        print("Verify your API keys and configuration")
        
    except Exception as e:
        print(f"Unexpected error: {e}")
        import traceback
        traceback.print_exc()

async def process_batch_queries(queries: list[str], config: dict):
    """Process multiple queries through the agent"""
    
    client = MultiServerMCPClient({
        "explorium": {
            "transport": "streamable_http",
            "url": "https://mcp.explorium.ai/mcp",
            "headers": {"api_key": config["explorium_api_key"]}
        }
    })
    
    async with client.session("explorium") as session:
        tools = await load_mcp_tools(session)
        model = ChatAnthropic(
            model="claude-3-5-sonnet-20241022",
            api_key=config["anthropic_api_key"]
        )
        
        # Build your graph here (same as before)
        # ...
        
        results = []
        for query in queries:
            initial_state = AgentState(
                messages=[HumanMessage(content=query)]
            )
            
            # Process each query
            async for event in graph.astream(initial_state):
                # Collect results
                pass
            
            results.append({"query": query, "response": "..."})
        
        return results

Using Explorium API key: dde4db9289...
Loaded 12 tools
Available tools:
  - match-business
  - fetch-businesses
  - fetch-businesses-statistics
  - fetch-businesses-events
  - enrich-business
  - match-prospects
  - fetch-prospects
  - fetch-prospects-events
  - fetch-prospects-statistics
  - enrich-prospects

AI: I'll help you find 10 product managers from Microsoft. Let me search for prospects with product manager roles at Microsoft.

Tool (match-business): {"response_context":{"correlation_id":"ac3bdd9b...","request_status":"success"...

AI: Now I'll search for product managers at Microsoft using the business ID I found.

Tool (fetch-prospects): {"response_context":{"correlation_id":"829dd0f9...","request_status":"success"...

AI: Here are 10 product managers from Microsoft:

## Microsoft Product Managers

1. **Demetrius M.** - Senior Product Manager, Azure Storage (United States)
2. **Alan Ross** - Partner Group Product Manager Manager, Cloud + AI Engineering (United States)
3. **Diego Rejtman** - Partner Group Product Manager, Threat Protection (United States)
4. **Mariam Keriakos** - Principal Product Manager Manager - M365 Copilot & Teams Platform (Germany)
5. **Drena Kusari** - Vice President of Product and General Manager (United States)
6. **Fatima Azzahra El Azzouzi** - Senior Product Manager - Data Security for AI (Canada)
7. **Jessica Peet** - Partner of Product Manager, Commercial Solution Areas CTO (United States)
8. **Seth Patton** - General Manager, Microsoft 365 Copilot, Product Marketing (United States)
9. **Morgan Webb** - Principal Group Manager Product Management | Security Engineering (United States)
10. **Raghav Jandhyala** - Partner General Manager, Head of Product - Microsoft Dynamics 365 AI (United States)

The search found 1,000 total product managers at Microsoft, so there are many more available if you need additional contacts.

import asyncio

async def parallel_tool_execution(tools, queries):
    """Execute multiple tool calls in parallel"""
    
    tasks = []
    for query in queries:
        task = asyncio.create_task(
            tools[0].ainvoke({"query": query})
        )
        tasks.append(task)
    
    results = await asyncio.gather(*tasks)
    return results