CData Connect AI Agent Chatbot

A production-ready chatbot built with the Claude Agent SDK that connects to the CData Connect AI MCP Server. Demonstrates how to build AI agents with stateful conversations, automatic context management, and native MCP tool integration.

• 🧠 Stateful Conversations: Automatic context management across multiple turns
• 🛠️ Built-in Tool Framework: Native MCP tool integration with proper lifecycle management
• 🎯 Production-Ready: Built on the same agent harness that powers Claude Code
• ⚡ Context Management: Automatic context compaction to prevent running out of context
• 🎛️ Advanced Controls: Fine-grained permissions, hooks, and behavioral customization
• 🔌 Connects to CData Connect AI MCP Server at https://mcp.cloud.cdata.com/mcp/
• 🔐 Basic authentication with email and personal access token
• 🤖 Uses Claude Sonnet 4.5 via Agent SDK
• 💬 Dynamic MCP tool loading and integration
• 🖥️ Interactive command-line interface
• ⚡ Async/await architecture for better performance

1. Install dependencies:

pip install -r requirements.txt

2. Set up environment variables:

cp .env.example .env
# Edit .env and add your credentials

3. Run the chatbot:

python3 agent_chatbot.py

• ANTHROPIC_API_KEY: Your Anthropic API key
• CDATA_EMAIL: Your CData account email
• CDATA_ACCESS_TOKEN: Your CData personal access token

export ANTHROPIC_API_KEY="your_api_key"
export CDATA_EMAIL="your_email@example.com"
export CDATA_ACCESS_TOKEN="your_personal_access_token"

Or create a .env file:

cp .env.example .env
# Edit .env with your credentials

python3 agent_chatbot.py

The chatbot creates a stateful session using ClaudeSDKClient, which:

• Maintains conversation context automatically
• Tracks tool usage across turns
• Manages memory and context compaction

from agent_chatbot import MCPAgentChatbot
import asyncio

async def main():
    chatbot = MCPAgentChatbot(
        mcp_server_url="https://mcp.cloud.cdata.com/mcp/",
        email="your_email@example.com",
        access_token="your_personal_access_token"
    )

    # Single query (creates new session)
    response = await chatbot.chat_once("What data sources are available?")
    print(response)

asyncio.run(main())

from agent_chatbot import MCPAgentChatbot
import asyncio

async def main():
    chatbot = MCPAgentChatbot(
        mcp_server_url="https://mcp.cloud.cdata.com/mcp/",
        email="your_email@example.com",
        access_token="your_personal_access_token"
    )

    # Create stateful session
    client = chatbot.create_session()

    # Multiple turns with context
    async with client:
        response1 = await chatbot.chat_session(client, "List my data sources")
        response2 = await chatbot.chat_session(client, "Tell me more about the first one")

asyncio.run(main())

1. MCP Tool Discovery: Fetches available tools from CData Connect AI
2. Tool Wrapping: Converts MCP tools to Agent SDK format using @tool decorator
3. MCP Server Creation: Creates SDK-compatible MCP server with create_sdk_mcp_server()
4. Agent Configuration: Sets up ClaudeAgentOptions with tools and permissions
5. Stateful Sessions: Uses ClaudeSDKClient for continuous conversations
6. Automatic Tool Calling: Agent SDK handles tool execution lifecycle automatically

@tool(
    name="data_source_query",
    description="Query a data source",
    input_schema={"type": "object", ...}
)
async def tool_handler(args):
    result = mcp_client.call_tool("data_source_query", args)
    return {"content": [{"type": "text", "text": json.dumps(result)}]}

mcp_server = create_sdk_mcp_server(
    name="cdata_connect",
    tools=[tool1, tool2, ...]
)

options = ClaudeAgentOptions(
    system_prompt="You are a helpful assistant...",
    mcp_servers={"cdata_connect": mcp_server},  # Dictionary of MCP servers
    permission_mode="bypassPermissions"  # Options: "default", "acceptEdits", "bypassPermissions", "plan"
)

client = ClaudeSDKClient(options=options)

# Maintains context automatically using async context manager
async with client:
    await client.query("First question")
    async for msg in client.receive_response():
        print(msg)

    await client.query("Follow-up question")
    async for msg in client.receive_response():
        print(msg)

• No manual conversation history tracking
• Automatic context compaction when approaching limits
• Built-in memory management

• Tools are first-class citizens
• Automatic tool calling lifecycle
• Better error handling and retries

• Session management
• Permission controls
• Hook system for customization

• Async/await throughout
• Streaming support
• Cleaner abstractions

• Verify your CData email is correct
• Ensure your personal access token is valid and not expired
• Check that you have access to the CData Connect AI service

• Verify connection to https://mcp.cloud.cdata.com/mcp/
• Check network connectivity and firewall settings
• Ensure MCP server is responding to tools/list requests

• Make sure claude-agent-sdk is installed: pip install claude-agent-sdk
• Check that you're using Python 3.8+
• Verify your ANTHROPIC_API_KEY is set correctly

• Claude Agent SDK Documentation
• Model Context Protocol (MCP)
• CData Connect AI