LangGraph

pip install mcp

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import httpx

# Pylar MCP server configuration
PYLAR_MCP_URL = "https://mcp.publish.pylar.ai/mcp"
PYLAR_BEARER_TOKEN = "YOUR_BEARER_TOKEN_HERE"

# Create HTTP client with authentication
async def create_pylar_client():
    async with httpx.AsyncClient() as client:
        headers = {
            "Authorization": f"Bearer {PYLAR_BEARER_TOKEN}",
            "Content-Type": "application/json"
        }
        
        # Connect to Pylar MCP server
        response = await client.post(
            f"{PYLAR_MCP_URL}/tools/list",
            headers=headers
        )
        
        if response.status_code == 200:
            tools = response.json()
            return tools
        else:
            raise Exception(f"Failed to connect: {response.status_code}")

# Use in your LangGraph agent
async def call_pylar_tool(tool_name, arguments):
    async with httpx.AsyncClient() as client:
        headers = {
            "Authorization": f"Bearer {PYLAR_BEARER_TOKEN}",
            "Content-Type": "application/json"
        }
        
        response = await client.post(
            f"{PYLAR_MCP_URL}/tools/call",
            headers=headers,
            json={
                "name": tool_name,
                "arguments": arguments
            }
        )
        
        return response.json()

from langgraph.graph import StateGraph, END
from langchain_core.tools import tool

# Define Pylar tools as LangChain tools
@tool
async def get_engagement_scores(event_type: str):
    """Get engagement scores for a given event type."""
    result = await call_pylar_tool(
        "fetch_engagement_scores_by_event_type",
        {"event_type": event_type}
    )
    return result

# Add to your agent's tool list
tools = [get_engagement_scores]

# Use in LangGraph workflow
def create_agent():
    workflow = StateGraph(AgentState)
    
    workflow.add_node("agent", your_agent_node)
    workflow.add_node("tools", execute_tools)
    
    workflow.set_entry_point("agent")
    workflow.add_edge("agent", "tools")
    workflow.add_edge("tools", END)
    
    return workflow.compile()

# Agent can call Pylar tools
result = await agent.invoke({
    "messages": [("user", "Get engagement scores for login events")]
})

# Tools are called automatically based on agent decisions

try:
    result = await call_pylar_tool(tool_name, arguments)
except Exception as e:
    # Handle errors gracefully
    logger.error(f"Pylar tool error: {e}")
    return {"error": str(e)}

from functools import lru_cache

@lru_cache(maxsize=100)
async def get_cached_data(tool_name, arguments):
    return await call_pylar_tool(tool_name, arguments)

from pydantic import BaseModel

class EngagementScoreArgs(BaseModel):
    event_type: str
    
async def get_engagement_scores(args: EngagementScoreArgs):
    return await call_pylar_tool(
        "fetch_engagement_scores_by_event_type",
        args.dict()
    )