IntegrationsMCP Server

Python SDK

Use the MCP SDK to integrate ConnectSafely into your Python applications

Use the MCP SDK to integrate this server into your Python applications.

Installation

Install the official MCP SDKs using pip:

pip install "mcp[cli]"

Or using uv:

uv add "mcp[cli]"

Python SDK

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
from urllib.parse import urlencode

# Construct server URL with authentication
base_url = "https://mcp.connectsafely.ai/"
params = {"apiKey": "YOUR_API_KEY"}
url = f"{base_url}?{urlencode(params)}"

async def main():
    # Connect to the server using HTTP client
    async with streamablehttp_client(url) as (read, write, _):
        async with ClientSession(read, write) as session:
            # Initialize the connection
            await session.initialize()

            # List available tools
            tools_result = await session.list_tools()
            print(f"Available tools: {', '.join([t.name for t in tools_result.tools])}")

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

Getting Your API Key

Get your API key from the MCP Setup page in your ConnectSafely dashboard.

Your MCP server URL format:

https://mcp.connectsafely.ai/?apiKey=YOUR_API_KEY

Usage Examples

Listing Available Tools

async def list_tools():
    async with streamablehttp_client(url) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # List all available tools
            tools_result = await session.list_tools()
            print("Available tools:")
            for tool in tools_result.tools:
                print(f"- {tool.name}: {tool.description}")

Calling MCP Tools

async def get_comments(post_id):
    async with streamablehttp_client(url) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # Call a specific tool
            result = await session.call_tool("get_linkedin_comments", {
                "postId": post_id,
                "limit": 10
            })
            print(f"Comments: {result}")

Error Handling

async def main():
    try:
        async with streamablehttp_client(url) as (read, write, _):
            async with ClientSession(read, write) as session:
                await session.initialize()
                tools_result = await session.list_tools()
                print("Connected successfully!")
    except Exception as error:
        print(f"Failed to connect to MCP server: {error}")

Complete Example

Here's a complete example of a Python application using ConnectSafely MCP server:

import asyncio
import os
from urllib.parse import urlencode
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async def main():
    # Construct server URL with authentication
    base_url = "https://mcp.connectsafely.ai/"
    api_key = os.getenv("CONNECTSAFELY_API_KEY", "YOUR_API_KEY")
    params = {"apiKey": api_key}
    url = f"{base_url}?{urlencode(params)}"

    try:
        # Connect to the server using HTTP client
        async with streamablehttp_client(url) as (read, write, _):
            async with ClientSession(read, write) as session:
                # Initialize the connection
                await session.initialize()
                print("Connected to ConnectSafely MCP server")

                # List available tools
                tools_result = await session.list_tools()
                print(f"\nAvailable tools ({len(tools_result.tools)}):")
                for tool in tools_result.tools:
                    print(f"- {tool.name}: {tool.description}")

                # Example: Get LinkedIn comments
                comments = await session.call_tool("get_linkedin_comments", {
                    "postId": "your-post-id",
                    "limit": 10
                })
                print(f"\nRecent comments: {comments}")

    except Exception as error:
        print(f"Error: {error}")

if __name__ == "__main__":
    asyncio.run(main())

Environment Variables

For better security, use environment variables to store your API key:

import os
from urllib.parse import urlencode

# Load from environment variable
api_key = os.getenv("CONNECTSAFELY_API_KEY")

if not api_key:
    raise ValueError("CONNECTSAFELY_API_KEY environment variable is required")

base_url = "https://mcp.connectsafely.ai/"
params = {"apiKey": api_key}
url = f"{base_url}?{urlencode(params)}"

Then set the environment variable:

export CONNECTSAFELY_API_KEY=your_api_key_here

Or in your .env file:

CONNECTSAFELY_API_KEY=your_api_key_here

Using python-dotenv:

from dotenv import load_dotenv
import os

# Load environment variables from .env file
load_dotenv()

api_key = os.getenv("CONNECTSAFELY_API_KEY")

Using with Different Async Frameworks

With asyncio

import asyncio

if __name__ == "__main__":
    asyncio.run(main())

With Trio

import trio
from mcp.client.streamable_http import streamablehttp_client

async def main():
    async with streamablehttp_client(url) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            # Your code here

if __name__ == "__main__":
    trio.run(main)

Security Best Practices

⚠️ Security Warning

Never commit your API keys to version control or expose them in public repositories.

Best Practices:

  • Use environment variables for API keys
  • Add .env to your .gitignore
  • Use different API keys for development and production
  • Regenerate keys if accidentally exposed
  • Monitor API usage regularly
  • Use python-dotenv for local development

Troubleshooting

Connection Failed

  • Verify your API key is correct
  • Ensure you have an active Pro or Agency subscription
  • Check your network connection

Module Not Found

Make sure you installed the MCP SDK with CLI support:

pip install "mcp[cli]"

Async Runtime Error

Ensure you're running the async code properly:

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

Learn More

TypeScript SDK

Use the MCP SDK to integrate ConnectSafely into your TypeScript applications

n8n Integration

Install and configure the ConnectSafely community node inside n8n.

On this page

InstallationPython SDKGetting Your API KeyUsage ExamplesListing Available ToolsCalling MCP ToolsError HandlingComplete ExampleEnvironment VariablesUsing with Different Async FrameworksWith asyncioWith TrioSecurity Best PracticesTroubleshootingConnection FailedModule Not FoundAsync Runtime ErrorLearn More