![]() |
VOOZH | about |
Claude Agent SDK is a powerful framework from Anthropic for building production-ready AI agents with stateful conversations, automatic context management, and native MCP (Model Context Protocol) tool integration. When combined with CData Connect AI, you can leverage Claude Agent SDK to build intelligent agents that interact with your ServiceNow data in real-time through natural language queries. This article outlines the process of connecting to ServiceNow using Connect AI and configuring a Claude Agent SDK application to interact with your ServiceNow data.
CData Connect AI offers a dedicated cloud-to-cloud interface for connecting to ServiceNow data. The CData Connect AI Remote MCP Server enables secure communication between Claude Agent SDK applications and ServiceNow. This allows your agents to read from and take actions on your ServiceNow data, all without the need for data replication to a natively supported database. With its inherent optimized data processing capabilities, CData Connect AI efficiently channels all supported SQL operations, including filters and JOINs, directly to ServiceNow. This leverages server-side processing to swiftly deliver the requested ServiceNow data.
In this article, we show how to configure a Claude Agent SDK application to conversationally explore (or Vibe Query) your data using natural language. With Connect AI you can build agents with access to live ServiceNow data, plus hundreds of other sources.
CData simplifies access and integration of live ServiceNow data. Our customers leverage CData connectivity to:
Many users access live ServiceNow data from preferred analytics tools like Tableau, Power BI, and Excel, and use CData solutions to integrate ServiceNow data with their database or data warehouse.
Connectivity to ServiceNow from Claude Agent SDK applications is made possible through CData Connect AI Remote MCP. To interact with ServiceNow data from your agent, we start by creating and configuring a ServiceNow connection in CData Connect AI.
ServiceNow uses the OAuth 2.0 authentication standard. To authenticate using OAuth, register an OAuth app with ServiceNow to obtain the OAuthClientId and OAuthClientSecret connection properties. In addition to the OAuth values, specify the Instance, Username, and Password connection properties.
See the "Getting Started" chapter in the help documentation for a guide on connecting to ServiceNow.
๐ Configuring a connection (Salesforce is shown)A Personal Access Token (PAT) is used to authenticate the connection to Connect AI from your Claude Agent SDK application. It is best practice to create a separate PAT for each service to maintain granularity of access.
With the connection configured and a PAT generated, we are ready to connect to ServiceNow data from your Claude Agent SDK application.
Follow these steps to configure your Claude Agent SDK application to connect to CData Connect AI. You can use our pre-built agent chatbot as a starting point, available at https://github.com/CDataSoftware/connect-ai-claude-agent, or follow the instructions below to create your own.
pip install claude-agent-sdk python-dotenv requests
mkdir my-cdata-agent cd my-cdata-agent
ANTHROPIC_API_KEY=your_anthropic_api_key [email protected] CDATA_ACCESS_TOKEN=your_personal_access_tokenReplace your_anthropic_api_key with your Anthropic API key, [email protected] with your Connect AI email address, and your_personal_access_token with the Personal Access Token created in Step 1.
With the setup for CData Connect AI completed, we are ready to build our agent application. We will walk through the key components needed to create a production-ready AI agent with stateful conversations and native MCP tool integration. For this example, we will create a single Python file (agent.py), which includes the MCP client, agent chatbot, and interactive interface.
To start, import the necessary libraries and load environment variables. We need the Claude Agent SDK for agent functionality, the requests library for HTTP communication with the MCP server, and standard Python libraries for async operations and authentication.
#!/usr/bin/env python3 import os import json import base64 import asyncio from typing import Optional, Dict, Any from functools import partial import requests from claude_agent_sdk import query, ClaudeSDKClient, ClaudeAgentOptions, tool, create_sdk_mcp_server from dotenv import load_dotenv # Load environment variables from .env file load_dotenv()
The MCPClient class handles communication with the CData Connect AI MCP server over HTTP. It manages authentication using Basic Auth (email and personal access token), sends JSON-RPC requests to the MCP server, and parses Server-Sent Events (SSE) responses. The client provides two key methods: list_tools() to discover available tools and call_tool() to execute MCP tools.
class MCPClient:
"""Client for interacting with CData Connect AI MCP server over HTTP."""
def __init__(self, server_url: str, email: Optional[str] = None, access_token: Optional[str] = None):
self.server_url = server_url.rstrip('/')
self.session = requests.Session()
# Set default headers for MCP JSON-RPC
self.session.headers.update({
'Content-Type': 'application/json',
'Accept': 'application/json, text/event-stream'
})
if email and access_token:
# Basic authentication: email:personal_access_token
credentials = f"{email}:{access_token}"
encoded_credentials = base64.b64encode(credentials.encode()).decode()
self.session.headers.update({
'Authorization': f'Basic {encoded_credentials}'
})
def _parse_sse_response(self, response_text: str) -> dict:
"""Parse Server-Sent Events (SSE) response."""
for line in response_text.split('\n'):
if line.startswith('data: '):
data_json = line[6:] # Remove 'data: ' prefix
return json.loads(data_json)
raise ValueError("No data found in SSE response")
def list_tools(self) -> list:
"""Get available tools from the MCP server."""
response = self.session.post(
self.server_url,
json={
"jsonrpc": "2.0",
"method": "tools/list",
"params": {},
"id": 1
}
)
response.raise_for_status()
result = self._parse_sse_response(response.text)
return result.get("result", {}).get("tools", [])
def call_tool(self, tool_name: str, arguments: dict) -> dict:
"""Call a tool on the MCP server."""
response = self.session.post(
self.server_url,
json={
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": tool_name,
"arguments": arguments
},
"id": 2
}
)
response.raise_for_status()
result = self._parse_sse_response(response.text)
return result.get("result", {})
The MCPAgentChatbot class is the core of our application. It connects the MCP client to the Claude Agent SDK, creating a production-ready AI agent with stateful conversations and automatic context management. The class performs several key functions during initialization: it creates an MCP client instance, discovers available tools from the CData Connect AI MCP server, wraps each MCP tool for use with the Agent SDK, and creates an SDK-compatible MCP server.
When the MCPAgentChatbot is initialized, it connects to the CData Connect AI MCP server and retrieves all available tools. These tools are then wrapped to be compatible with the Claude Agent SDK's tool framework.
class MCPAgentChatbot:
"""
Production-ready AI agent chatbot for CData Connect AI.
Uses Claude Agent SDK for stateful conversations and automatic context management.
"""
def __init__(self, mcp_server_url: str, email: Optional[str] = None, access_token: Optional[str] = None):
self.mcp_client = MCPClient(mcp_server_url, email, access_token)
# Load available tools from MCP server
print("Connecting to CData Connect AI MCP server...")
self.mcp_tools_list = self.mcp_client.list_tools()
print(f"โ Loaded {len(self.mcp_tools_list)} tools from MCP server")
# Create custom MCP tool wrappers for the Agent SDK
self.agent_tools = self._create_agent_tools()
# Create MCP server for Agent SDK
self.mcp_server = create_sdk_mcp_server(
name="cdata_connect",
tools=self.agent_tools
)
The _create_agent_tools() method transforms MCP tool definitions into Agent SDK-compatible tools. For each tool discovered from the MCP server, it creates a wrapper using the @tool decorator with the tool's name, description, and input schema. The _tool_handler() method is called whenever a tool is executed - it forwards the request to the MCP server and formats the response for the Agent SDK.
async def _tool_handler(self, tool_name: str, args: Dict[str, Any]) -> Dict[str, Any]:
"""Call the MCP tool and return results."""
result = self.mcp_client.call_tool(tool_name, args)
return {
"content": [{
"type": "text",
"text": json.dumps(result, indent=2)
}]
}
def _create_agent_tools(self) -> list:
"""Create Agent SDK tool wrappers for MCP tools."""
agent_tools = []
for tool_info in self.mcp_tools_list:
tool_name = tool_info.get("name")
tool_description = tool_info.get("description", "")
tool_schema = tool_info.get("inputSchema", {})
agent_tool = tool(
name=tool_name,
description=tool_description,
input_schema=tool_schema
)(partial(self._tool_handler, tool_name))
agent_tools.append(agent_tool)
return agent_tools
One of the key features of the Claude Agent SDK is its ability to maintain stateful conversations with automatic context management. The create_session() method creates a ClaudeSDKClient instance configured with a system prompt, access to the MCP tools, and permission settings. The chat_session() method sends a user message and retrieves the agent's response, maintaining conversation context automatically.
def create_session(self) -> ClaudeSDKClient:
"""Create a stateful conversation session."""
options = ClaudeAgentOptions(
system_prompt="You are a helpful assistant with access to ServiceNow data through CData Connect AI.",
mcp_servers={"cdata_connect": self.mcp_server},
permission_mode="bypassPermissions"
)
return ClaudeSDKClient(options=options)
async def chat_session(self, client: ClaudeSDKClient, user_message: str) -> str:
"""Send a message in a stateful session."""
await client.query(user_message)
async for message in client.receive_response():
if hasattr(message, 'result'):
return str(message.result)
return ""
The main() function brings everything together to create an interactive chatbot. It loads credentials from environment variables, initializes the MCPAgentChatbot, creates a stateful session, and starts an interactive loop where users can chat with the agent. The session is managed using Python's async context manager (async with), which ensures proper cleanup of resources and API connections.
async def main():
"""Run the agent in interactive mode."""
MCP_SERVER_URL = "https://mcp.cloud.cdata.com/mcp/"
CDATA_EMAIL = os.environ.get("CDATA_EMAIL")
CDATA_ACCESS_TOKEN = os.environ.get("CDATA_ACCESS_TOKEN")
print("=== CData Connect AI Agent ===\n")
chatbot = MCPAgentChatbot(MCP_SERVER_URL, CDATA_EMAIL, CDATA_ACCESS_TOKEN)
client = chatbot.create_session()
async with client:
while True:
user_input = input("You: ").strip()
if user_input.lower() in ['quit', 'exit', 'q']:
break
response = await chatbot.chat_session(client, user_input)
print(f"\nAssistant: {response}\n")
if __name__ == "__main__":
asyncio.run(main())
python agent.py
With your Claude Agent SDK application configured and connected to CData Connect AI, you can now build sophisticated agents that interact with your ServiceNow data using natural language. The MCP integration provides your agents with powerful data access capabilities.
Your Claude Agent SDK application has access to the following CData Connect AI MCP tools:
The Claude Agent SDK provides several production-ready features that make it ideal for building AI agents:
Here are some examples of what your Claude Agent SDK applications can do with live ServiceNow data access:
Once your agent is running, you can interact with it through natural language queries. For example:
Your Claude Agent SDK application will automatically translate these natural language queries into appropriate SQL queries and execute them against your ServiceNow data through the CData Connect AI MCP Server, providing real-time insights without requiring users to write complex SQL or understand the underlying data structure.
You can also use the Claude Agent SDK programmatically for one-off queries or integration into larger applications:
from claude_agent_sdk import query, ClaudeAgentOptions
async def single_query():
options = ClaudeAgentOptions(
system_prompt="You are a data analyst assistant.",
mcp_servers={"cdata_connect": mcp_server},
permission_mode="bypassPermissions"
)
response = await query(
prompt="What data sources are available?",
options=options
)
async for message in response:
if hasattr(message, 'result'):
print(message.result)
async def conversation(): client = chatbot.create_session() async with client: # First query response1 = await chatbot.chat_session(client, "List my data sources") print(response1) # Follow-up query with context response2 = await chatbot.chat_session(client, "Tell me more about the first one") print(response2)
Claude Agent SDK offers different permission modes for tool execution:
To get live data access to hundreds of SaaS, Big Data, and NoSQL sources directly from your Claude Agent SDK applications, try CData Connect AI today!
Learn more about CData Connect AI or sign up for free trial access:
Free Trial