openai-integration.py•8.83 kB
#!/usr/bin/env python3
"""
Example integration of ValueRouter MCP server with OpenAI function calling.
This demonstrates how to bridge the MCP server tools with OpenAI's API.
"""
import subprocess
import json
import os
from typing import Dict, Any, List
import openai
class ValueRouterMCP:
"""Bridge between OpenAI function calling and ValueRouter MCP server."""
def __init__(self):
self.client = openai.OpenAI()
self.mcp_process = None
self.start_mcp_server()
def start_mcp_server(self):
"""Start the ValueRouter MCP server process."""
env = os.environ.copy()
env.update({
'ETHEREUM_RPC_URL': 'https://mainnet.infura.io/v3/YOUR_INFURA_KEY',
'ARBITRUM_RPC_URL': 'https://arb1.arbitrum.io/rpc',
'SOLANA_RPC_URL': 'https://api.mainnet-beta.solana.com',
# Add other RPC URLs as needed
})
self.mcp_process = subprocess.Popen(
['npx', '@valuerouter/mcp-server'],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
env=env
)
def call_mcp_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Call a tool on the MCP server."""
request = {
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": tool_name,
"arguments": arguments
}
}
# Send request to MCP server
self.mcp_process.stdin.write(json.dumps(request) + '\n')
self.mcp_process.stdin.flush()
# Read response
response_line = self.mcp_process.stdout.readline()
response = json.loads(response_line)
if 'error' in response:
raise Exception(f"MCP Error: {response['error']}")
return response.get('result', {})
def get_openai_function_definitions(self) -> List[Dict[str, Any]]:
"""Get OpenAI function definitions for all ValueRouter tools."""
return [
{
"name": "get_supported_chains",
"description": "Get all supported blockchain chains for USDC bridging",
"parameters": {
"type": "object",
"properties": {
"includeTestnets": {
"type": "boolean",
"description": "Whether to include testnet chains"
}
}
}
},
{
"name": "get_bridge_quote",
"description": "Get a quote for bridging USDC between blockchain chains",
"parameters": {
"type": "object",
"properties": {
"fromChainId": {
"type": "string",
"description": "Source chain ID (e.g., '1' for Ethereum)"
},
"toChainId": {
"type": "string",
"description": "Destination chain ID (e.g., '42161' for Arbitrum)"
},
"amount": {
"type": "string",
"description": "Amount to bridge in smallest unit (e.g., '1000000' for 1 USDC)"
}
},
"required": ["fromChainId", "toChainId", "amount"]
}
},
{
"name": "get_user_balance",
"description": "Get user's token balance on a specific blockchain",
"parameters": {
"type": "object",
"properties": {
"chainId": {
"type": "string",
"description": "Chain ID to check balance on"
},
"tokenAddress": {
"type": "string",
"description": "Token contract address"
},
"userAddress": {
"type": "string",
"description": "User's wallet address"
}
},
"required": ["chainId", "tokenAddress", "userAddress"]
}
},
{
"name": "get_transaction_status",
"description": "Get the status of a bridge transaction",
"parameters": {
"type": "object",
"properties": {
"transactionHash": {
"type": "string",
"description": "Transaction hash to check"
},
"fromChainId": {
"type": "string",
"description": "Source chain ID"
},
"toChainId": {
"type": "string",
"description": "Destination chain ID"
}
},
"required": ["transactionHash", "fromChainId", "toChainId"]
}
}
]
def bridge_usdc_with_ai(self, user_message: str) -> str:
"""Use OpenAI to understand user intent and execute bridge operations."""
system_prompt = """
You are a cross-chain bridge assistant powered by ValueRouter. You can help users:
1. Check supported blockchain networks
2. Get quotes for bridging USDC between chains
3. Check token balances across different blockchains
4. Track transaction status
Always be helpful and explain what you're doing. When bridging, make sure to:
- Show the user the quote with fees
- Explain the estimated time
- Confirm the transaction details before proceeding
Available chains include Ethereum (1), Arbitrum (42161), Optimism (10),
Polygon (137), Avalanche (43114), Base (8453), Solana (solana), and Sui (sui).
"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message}
]
response = self.client.chat.completions.create(
model="gpt-4",
messages=messages,
functions=self.get_openai_function_definitions(),
function_call="auto"
)
message = response.choices[0].message
if message.function_call:
# Execute the function call via MCP
function_name = message.function_call.name
function_args = json.loads(message.function_call.arguments)
try:
result = self.call_mcp_tool(function_name, function_args)
# Send result back to OpenAI for interpretation
messages.append({
"role": "assistant",
"content": None,
"function_call": {
"name": function_name,
"arguments": message.function_call.arguments
}
})
messages.append({
"role": "function",
"name": function_name,
"content": json.dumps(result)
})
# Get final response from OpenAI
final_response = self.client.chat.completions.create(
model="gpt-4",
messages=messages
)
return final_response.choices[0].message.content
except Exception as e:
return f"Error executing bridge operation: {str(e)}"
return message.content
# Example usage
if __name__ == "__main__":
# Initialize the bridge
bridge = ValueRouterMCP()
# Example interactions
examples = [
"What chains are supported for USDC bridging?",
"I want to bridge 100 USDC from Ethereum to Arbitrum",
"Check my USDC balance on Ethereum for address 0x742d35Cc6634C0532925a3b8D45c07F9624f8b4c",
"What's the status of transaction hash 0x123abc...?"
]
for example in examples:
print(f"\n--- User: {example} ---")
response = bridge.bridge_usdc_with_ai(example)
print(f"Assistant: {response}")