import os
import asyncio
import base64
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
# Configure StackOne account
STACKONE_ACCOUNT_ID = "<account_id>" # Your StackOne account ID
# Encode API key for Basic auth
auth_token = base64.b64encode(
f"{os.getenv('STACKONE_API_KEY')}:".encode()
).decode()
async def main():
# Connect to StackOne MCP server
mcp_client = MultiServerMCPClient({
"stackone": {
"url": "https://api.stackone.com/mcp",
"transport": "streamable_http",
"headers": {
"Authorization": f"Basic {auth_token}",
"x-account-id": STACKONE_ACCOUNT_ID,
"Content-Type": "application/json",
"Accept": "application/json,text/event-stream", # Required by MCP spec
"MCP-Protocol-Version": "2025-06-18"
}
}
})
# Get StackOne tools
tools = await mcp_client.get_tools()
# Create agent with StackOne tools
llm = ChatOpenAI(model="gpt-5")
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant with access to data from connected platforms."),
("human", "{input}"),
("placeholder", "{agent_scratchpad}")
])
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)
# Run agent
result = agent_executor.invoke({"input": "List Salesforce accounts"})
print(result["output"])
asyncio.run(main())
MCP
A2A