Prerequisites
- Tool Pack ID: ID of the tool pack you wish to expose to your user at the time of session. This ID can be retrieved within the dashboard or via API using the GET /tool-packs call.
- Registered User ID: ID of the user who will be interacting with your agent at the time of session. This ID can be retrieved within the dashboard or via API when first creating a Registered User.
- Production Access Key: Your Merge Agent Handler API Key used to authenticate requests.
NOTE: If you are using a Test Registered User, you must also use a test production access key. If you are using a Production Registered User, you must use a production access key.
MCP Entry URL
The MCP Entry URL is a combination of Tool Pack ID and Registered User ID, and looks like this:Copy
https://ah-api.merge.dev/api/v1/tool-packs/<tool-pack-id>/registered-users/<registered-user-id>/mcp
Implementation of Merge Agent Handler MCP Server
Config File
Many applications like Claude Desktop, Cursor, or Windsurf support MCP servers out of the box and will handle the connection to Merge Agent Handler.Copy
{
"mcpServers": {
"agent-handler": {
"command": "npx",
"args": [
"-y",
"mcp-remote@latest",
"https://ah-api.merge.dev/api/v1/tool-packs/<tool-pack-id>/registered-users/<registered-user-id>/mcp",
"--header",
"Authorization: Bearer ${AUTH_TOKEN}"
],
"env": {
"AUTH_TOKEN": "<ah-production-access-key>"
}
}
}
}
Implement using MCP’s Python SDK
For MCP Client builders, the following code snippet can be used to connect to Merge Agent Handler.Copy
import asyncio
import logging
import sys
# Set up basic logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("mcp-client")
async def connect_to_mcp():
"""Connect to an MCP server and use its tools"""
from mcp.client.streamable_http import streamablehttp_client
from mcp import ClientSession
# Custom headers for authentication and other requirements
headers = {
"Authorization": "Bearer <auth_token>",
"Mcp-Session-Id": "session-" + str(asyncio.current_task().get_name())
}
# MCP server URL
server_url = "https://ah-api.merge.dev/api/v1/tool-packs/<tool_pack_id>/registered-users/<registered_user_id>/mcp"
logger.info(f"Connecting to MCP server: {server_url}")
try:
# Connect to the MCP server with streamablehttp_client
async with streamablehttp_client(server_url, headers=headers) as (read_stream, write_stream, _):
logger.info("Connection established, creating ClientSession")
# Create a session using the client streams
async with ClientSession(read_stream, write_stream) as session:
# Initialize the connection with the correct protocol version
logger.info("Initializing session with protocol version 2024-11-05")
await session.initialize()
# List available tools
logger.info("Listing tools")
tools_result = await session.list_tools()
logger.info("\nMCP Tools:")
# Debug the format of the returned tools
logger.info(tools_result)
# # Example: Call a specific tool
# tool_name = "<tool_name>" # Replace with an actual tool from your server
# try:
# tool_result = await session.call_tool(
# tool_name,
# {"<tool-args>": {}}
# )
# logger.info(f"\nResults from {tool_name}:")
# logger.info(tool_result)
# except Exception as e:
# logger.error(f"Error calling tool {tool_name}: {e}")
except Exception as e:
logger.exception(f"Error in MCP client: {e}")
async def main():
logger.info("Starting MCP client")
await connect_to_mcp()
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
logger.info("Program terminated by user")
sys.exit(0)
except Exception as e:
logger.exception(f"Unhandled exception: {e}")
sys.exit(1)
Implement using a custom Python client
Copy
import json
import uuid
import asyncio
import aiohttp
class MCPClient:
"""
Model Context Protocol (MCP) client for interacting with Merge Agent Handler MCP Server
This client implements the JSON-RPC 2.0 protocol used by the MCP server.
It supports initializing a session, listing available tools, and executing tools.
Uses async methods to match the server's async implementation.
"""
def __init__(self, base_url, auth_token=None):
"""
Initialize MCP client
Args:
base_url (str): The base URL for the MCP server
auth_token (str, optional): Bearer token for authentication
"""
self.base_url = base_url
self.session_id = str(uuid.uuid4())
self.request_id = 1
self.headers = {
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream",
"Mcp-Session-Id": self.session_id
}
if auth_token:
self.headers["Authorization"] = f"Bearer {auth_token}"
self.session = None
async def _ensure_session(self):
"""
Ensure that an aiohttp ClientSession exists
"""
if self.session is None or self.session.closed:
self.session = aiohttp.ClientSession()
return self.session
async def _send_request(self, method, params=None):
"""
Send an async JSON-RPC request to the MCP server
Args:
method (str): JSON-RPC method name
params (dict, optional): Parameters for the request
Returns:
dict: Response from the server
"""
payload = {
"jsonrpc": "2.0",
"id": self.request_id,
"method": method
}
if params:
payload["params"] = params
session = await self._ensure_session()
try:
async with session.post(
self.base_url,
headers=self.headers,
json=payload
) as response:
# Update session ID if provided in the response
if "Mcp-Session-Id" in response.headers:
self.session_id = response.headers["Mcp-Session-Id"]
self.headers["Mcp-Session-Id"] = self.session_id
# Increment request ID for next call
self.request_id += 1
# Check if the request was successful
response.raise_for_status()
# Try to parse as JSON first
try:
return await response.json()
except aiohttp.ContentTypeError:
# If not JSON, return the text content
return {"text": await response.text()}
except aiohttp.ClientError as e:
return {"error": f"Request failed: {str(e)}"}
async def initialize(self, protocol_version="2024-11-05"):
"""
Initialize a session with the MCP server
Args:
protocol_version (str): Version of the MCP protocol to use
Returns:
dict: Server capabilities and information
"""
params = {"protocolVersion": protocol_version}
return await self._send_request("initialize", params)
async def list_tools(self):
"""
List available tools from the MCP server
Returns:
dict: Available tools and their metadata
"""
return await self._send_request("tools/list")
async def call_tool(self, tool_name, arguments=None):
"""
Execute a tool on the MCP server
Args:
tool_name (str): Name of the tool to call
arguments (dict, optional): Arguments to pass to the tool
Returns:
dict: Result of the tool execution
"""
params = {
"name": tool_name,
"arguments": arguments or {}
}
return await self._send_request("tools/call", params)
async def close(self):
"""
Close the aiohttp session
"""
if self.session and not self.session.closed:
await self.session.close()
# Example usage
async def main():
# URL format: https://<host>/api/v1/tool-packs/<tool_pack_id>/registered-users/<user_id>/mcp
client = MCPClient(
base_url="https://ah-api.merge.dev/api/v1/tool-packs/<tool_pack_id>/registered-users/<user_id>/mcp",
auth_token="<ah-production-access-key>"
)
try:
# Initialize the session
init_result = await client.initialize()
print("\nMCP Initialization Response:")
print(json.dumps(init_result, indent=2))
# List available tools
tools_result = await client.list_tools()
print("\nMCP Tools List Response:")
print(json.dumps(tools_result, indent=2))
# Call a specific tool
# tool_result = await client.call_tool(
# tool_name="<tool_name>",
# arguments={<arguments>}
# )
# print("\nMCP Tool Call Response:")
# print(json.dumps(tool_result, indent=2))
finally:
# Always close the client session
await client.close()
if __name__ == "__main__":
asyncio.run(main())