Connect AI Clients to Actian MCP Server¶

"""Actian MCP Server — Python client example."""

import asyncio
import json
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport


async def main():
    # Replace with your Actian MCP Server URL
    server_url = "http://localhost:8000/mcp"

    transport = StreamableHttpTransport(url=server_url)

    async with Client(transport, timeout=60) as client:

        # 1. Discover available tools and their parameters
        tools = await client.list_tools()
        print("Available tools:")
        for tool in tools:
            print(f"  - {tool.name}")

        # Print parameter schema for a specific tool
        for tool in tools:
            if tool.name == "execute_query":
                print(f"\nexecute_query parameters:")
                print(json.dumps(tool.model_dump()["parameters"], indent=2))

        # 2. List tables in the connected database
        result = await client.call_tool("list_tables", {})
        print(f"\nTables:\n{result.content[0].text}")

        # 3. Describe a specific table
        # Replace "customers" with a table from your database
        # For Zen, use {"table": "customers"} instead
        result = await client.call_tool(
            "describe_table", {"table_name": "customers"}
        )
        print(f"\nTable structure:\n{result.content[0].text}")

        # 4. Execute a read-only SQL query
        # Replace table and column names to match your schema
        # For Zen, use {"sql": "..."} instead of {"query": "..."}
        result = await client.call_tool(
            "execute_query",
            {"query": "SELECT name, email FROM customers WHERE status = 'active'"},
        )
        print(f"\nQuery results:\n{result.content[0].text}")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except Exception as e:
        print(f"Error: {e}")

"""Actian MCP Server — Python client with OAuth and TLS."""

import asyncio
import ssl
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport
import httpx


async def main():
    server_url = "https://mcp.example.com:8000/mcp"

    # Trust a self-signed certificate (omit for publicly trusted certs)
    ssl_ctx = ssl.create_default_context(cafile="/path/to/server.crt")

    transport = StreamableHttpTransport(
        url=server_url,
        auth="oauth",
        httpx_client_factory=lambda **kwargs: httpx.AsyncClient(
            verify=ssl_ctx, **kwargs
        ),
    )

    async with Client(transport, timeout=120) as client:
        tools = await client.list_tools()
        print(f"Connected — {len(tools)} tools available")

        # Verify the authenticated database user
        # For Zen, use {"sql": "..."} instead of {"query": "..."}
        result = await client.call_tool(
            "execute_query", {"query": "SELECT CURRENT_USER"}
        )
        print(f"Current user: {result.content[0].text}")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except Exception as e:
        print(f"Error: {e}")