MCP Client Guide

Table of Contents

1. Prerequisites
2. Installation
3. Authentication Methods
4. Basic Usage Patterns
5. Working with Tools
6. Configuration Options
7. Properties and Methods
8. Error Handling
9. Advanced Usage
10. Best Practices
11. Complete Examples

Prerequisites

Before using the MCP client, ensure you have completed the following setup:

1. Register an Application

Applications act as the identity for your agent. Registration creates the credentials that allow your agent to access tools on behalf of users.

# Register your application (requires authentication)
afctl auth login
afctl applications register \
    --app-id my-email-agent \
    --scopes google:gmail.send,google:gmail.readonly,slack:chat:write \
    --display-name "My Email Agent"

# Optional: provide IdP credentials for SSO/token exchange
afctl applications register \
    --app-id my-email-agent \
    --scopes google:gmail.send,google:gmail.readonly,slack:chat:write \
    --display-name "My Email Agent" \
    --idp-client-id <idp-client-id> \
    --idp-client-secret <idp-client-secret>copy

If you do not pass the IdP flags, afctl prompts you for IdP credentials during registration. Press Enter to skip them if you are not using SSO, or pass --skip-idp to skip the prompt entirely.

After registration, you'll have:

• app_id: Your application identifier (e.g., org-abc123_my-email-agent)
• app_secret: Your application client secret (e.g., sk_live_xyz789...)

These credentials are saved to ~/.af/applications/{app_id}.json.

2. User Authentication

Users must authenticate before the MCP client can access their tools. The authentication method depends on your deployment:

3. Tool Connections

Users must connect the tools they want to use. This is done via:

• Dashboard UI: Users connect tools in the Agentic Fabriq dashboard
• CLI: afctl tools add and afctl tools connect

The MCP client can only access tools that:

1. The user has connected
2. The application has scopes for

Installation

pip install agentic-fabriq-sdkcopy

Import the client:

from af_sdk import MCPClientcopy

Authentication Methods

The MCP client supports three authentication methods, each suited for different deployment scenarios.

CLI Method (method="cli")

Use case: Local development, testing, CLI scripts, or any scenario where the user has already authenticated via afctl auth login.

How it works: The client reads stored credentials from ~/.af/credentials.json (created by afctl auth login) and exchanges them for an AF token.

Required parameters: app_id, app_secret

# User must first run: afctl auth login

async with MCPClient(
    method="cli",
    app_id="org-abc123_my-agent",
    app_secret="sk_live_xyz789..."
) as client:
    tools = await client.list_tools()copy

Keycloak Method (method="keycloak")

Use case: Web applications or services that authenticate users directly through Keycloak (Agentic Fabriq's identity provider).

Required parameters: app_id, app_secret, keycloak_token

# Token obtained from your Keycloak OAuth flow
keycloak_token = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9..."

async with MCPClient(
    method="keycloak",
    app_id="org-abc123_my-agent",
    app_secret="sk_live_xyz789...",
    keycloak_token=keycloak_token
) as client:
    result = await client.call_tool("slack_post_message", {
        "channel": "#general",
        "text": "Hello from my web app!"
    })copy

IdP Method (method="idp")

Use case: Enterprise organizations using external identity providers like Okta for SSO.

Required parameters: app_id, app_secret, external_token

Optional parameters: org_url (auto-detected from app if not provided, used to identify organization that user is in)

# Token obtained from Okta OAuth flow
okta_token = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9..."

async with MCPClient(
    method="idp",
    app_id="org-abc123_my-agent",
    app_secret="sk_live_xyz789...",
    external_token=okta_token,
    org_url="mycompany.com"  # Optional
) as client:
    result = await client.call_tool("notion_search", {
        "query": "Q4 planning"
    })copy

Basic Usage Patterns

Async Context Manager (Recommended)

The async context manager automatically handles connection and cleanup:

from af_sdk import MCPClient

async with MCPClient(
    method="cli",
    app_id="org-abc123_my-agent",
    app_secret="sk_live_xyz789..."
) as client:
    # Client is connected and authenticated
    tools = await client.list_tools()
    result = await client.call_tool("google_gmail_list_messages", {"max_results": 10})
    # Connection automatically closed on exitcopy

Sync Context Manager

For synchronous code, use the sync context manager with _sync method variants:

from af_sdk import MCPClient

with MCPClient(
    method="cli",
    app_id="org-abc123_my-agent",
    app_secret="sk_live_xyz789..."
) as client:
    tools = client.list_tools_sync()
    result = client.call_tool_sync("slack_post_message", {
        "channel": "#general",
        "text": "Hello from sync code!"
    })copy

Manual Connection Management

For fine-grained control over the connection lifecycle:

from af_sdk import MCPClient

client = MCPClient(
    method="cli",
    app_id="org-abc123_my-agent",
    app_secret="sk_live_xyz789..."
)

try:
    # Explicitly connect
    await client.connect()
    
    # Use the client
    tools = await client.list_tools()
    result = await client.call_tool("slack_post_message", {
        "channel": "#test",
        "text": "Hello!"
    })
finally:
    # Always disconnect
    await client.disconnect()copy

Working with Tools

Listing Available Tools

The list_tools() method fetches all tools available to the current user based on their connections and the application's scopes:

async with MCPClient(method="cli", app_id="...", app_secret="...") as client:
    tools = await client.list_tools()
    
    print(f"Found {len(tools)} tools")
    for tool in tools:
        print(f"\nTool: {tool['name']}")
        print(f"  Description: {tool['description']}")
        print(f"  Parameters: {tool['inputSchema']}")copy

Each tool definition includes:

• name: The tool identifier (e.g., google_gmail_send_email)
• description: Human-readable description of what the tool does
• inputSchema: JSON Schema defining the tool's parameters

Accessing Cached Tools

After calling list_tools(), tools are cached for quick access:

async with MCPClient(method="cli", app_id="...", app_secret="...") as client:
    await client.list_tools()  # Populates cache
    
    # Quick access to tool names
    print(client.tool_names)
    # ['google_gmail_list_messages', 'google_gmail_send_email', 'slack_post_message', ...]
    
    # Check if a specific tool is available
    if client.has_tool("google_gmail_send_email"):
        print("Gmail send is available!")
    
    # Get full tool definition
    tool = client.get_tool("slack_post_message")
    if tool:
        print(f"Slack tool schema: {tool['inputSchema']}")
    
    # Get all cached tools (no network call)
    all_tools = client.get_tools()copy

Understanding Tool Schemas

Each tool has an inputSchema that defines its parameters:

tool = client.get_tool("google_gmail_send_email")
print(tool['inputSchema'])copy

Example schema:

{
  "type": "object",
  "properties": {
    "to": {
      "type": "string",
      "description": "Recipient email address"
    },
    "subject": {
      "type": "string",
      "description": "Email subject line"
    },
    "body": {
      "type": "string",
      "description": "Email body content"
    },
    "cc": {
      "type": "string",
      "description": "CC recipients (optional)"
    }
  },
  "required": ["to", "subject", "body"]
}copy

Calling Tools

Use call_tool() with the tool name and a dictionary of arguments:

result = await client.call_tool("tool_name", {
    "param1": "value1",
    "param2": "value2"
})copy

The result format depends on the specific tool. Most tools return dictionaries or lists.

Configuration Options

Properties and Methods

Properties

Connection Methods

Tool Methods

Error Handling

The SDK provides specific exception types for different error scenarios:

from af_sdk import MCPClient
from af_sdk.exceptions import (
    AuthenticationError,
    MCPConnectionError,
    MCPError
)

try:
    async with MCPClient(
        method="cli",
        app_id="org-abc123_my-agent",
        app_secret="sk_live_xyz789..."
    ) as client:
        result = await client.call_tool("google_gmail_send_email", {
            "to": "user@example.com",
            "subject": "Test",
            "body": "Hello"
        })
        
except AuthenticationError as e:
    # Authentication failed
    # Causes: Token expired, invalid credentials, not logged in, invalid app_id/secret
    print(f"Authentication failed: {e}")
    print("Suggestion: Run 'afctl auth login' or check your app credentials")
    
except MCPConnectionError as e:
    # Connection failed
    # Causes: Network issues, server unreachable, timeout, connection not established
    print(f"Connection failed: {e}")
    print("Suggestion: Check network connectivity and MCP server URL")
    
except MCPError as e:
    # MCP server returned an error
    # Causes: Invalid tool name, bad parameters, tool execution failure, permission denied
    print(f"MCP error: {e.message}")
    print(f"Error code: {e.code}")
    print(f"Additional details: {e.data}")copy

Exception Reference

Advanced Usage

Custom Server URLs

For self-hosted deployments or staging environments:

async with MCPClient(
    method="cli",
    app_id="org-abc123_my-agent",
    app_secret="sk_live_xyz789...",
    mcp_url="https://mcp.mycompany.com/mcp",
    gateway_url="https://api.mycompany.com"
) as client:
    tools = await client.list_tools()copy

Inspecting Token Information

After connecting, you can inspect the authenticated token:

async with MCPClient(method="cli", app_id="...", app_secret="...") as client:
    # Token info is available after connect
    token_info = client.token_info
    
    print(f"User ID: {token_info.user_id}")
    print(f"App ID: {token_info.app_id}")
    print(f"Expires in: {token_info.expires_in} seconds")
    print(f"Token type: {token_info.token_type}")copy

Reusing Connections

For applications that make many tool calls, keep the connection open:

class MyAgent:
    def __init__(self, app_id: str, app_secret: str):
        self.client = MCPClient(
            method="cli",
            app_id=app_id,
            app_secret=app_secret
        )
    
    async def start(self):
        await self.client.connect()
        await self.client.list_tools()
    
    async def stop(self):
        await self.client.disconnect()
    
    async def send_email(self, to: str, subject: str, body: str):
        return await self.client.call_tool("google_gmail_send_email", {
            "to": to,
            "subject": subject,
            "body": body
        })
    
    async def post_slack(self, channel: str, text: str):
        return await self.client.call_tool("slack_post_message", {
            "channel": channel,
            "text": text
        })

# Usage
agent = MyAgent("org-abc123_my-agent", "sk_live_xyz789...")
await agent.start()
try:
    await agent.send_email("user@example.com", "Hello", "World")
    await agent.post_slack("#general", "Email sent!")
finally:
    await agent.stop()copy

Custom Timeout

For long-running tool operations:

async with MCPClient(
    method="cli",
    app_id="org-abc123_my-agent",
    app_secret="sk_live_xyz789...",
    timeout=120.0  # 2 minute timeout
) as client:
    # Long-running operation
    result = await client.call_tool("some_slow_tool", {"large_data": "..."})copy

Best Practices

1. Always Use Context Managers

Context managers ensure proper cleanup even if errors occur:

# Good
async with MCPClient(...) as client:
    await client.call_tool(...)

# Avoid (requires manual cleanup)
client = MCPClient(...)
await client.connect()
# If an error occurs here, disconnect() may not be called
await client.call_tool(...)
await client.disconnect()copy

2. Check Tool Availability Before Calling

async with MCPClient(...) as client:
    await client.list_tools()
    
    if not client.has_tool("google_gmail_send_email"):
        print("Gmail not available - user may need to connect it")
        return
    
    await client.call_tool("google_gmail_send_email", {...})copy

3. Handle Errors Gracefully

from af_sdk.exceptions import AuthenticationError, MCPError

async def send_notification(client, message):
    try:
        await client.call_tool("slack_post_message", {
            "channel": "#alerts",
            "text": message
        })
        return True
    except MCPError as e:
        logger.error(f"Failed to send Slack notification: {e.message}")
        return Falsecopy

4. Cache Tool Lists When Possible

If your application makes many calls, avoid repeated list_tools() calls:

async with MCPClient(...) as client:
    # Fetch once
    await client.list_tools()
    
    # Use cached data for subsequent checks
    if client.has_tool("google_gmail_send_email"):
        await client.call_tool("google_gmail_send_email", {...})
    
    if client.has_tool("slack_post_message"):
        await client.call_tool("slack_post_message", {...})copy

5. Use Appropriate Auth Method

Complete Examples

Email Assistant Agent

import asyncio
from af_sdk import MCPClient
from af_sdk.exceptions import AuthenticationError, MCPError

async def email_assistant():
    """An agent that summarizes unread emails and posts to Slack."""
    
    try:
        async with MCPClient(
            method="cli",
            app_id="org-abc123_email-assistant",
            app_secret="sk_live_xyz789..."
        ) as client:
            # Check what tools are available
            await client.list_tools()
            
            has_gmail = client.has_tool("google_gmail_list_messages")
            has_slack = client.has_tool("slack_post_message")
            
            if not has_gmail:
                print("Error: Gmail not connected")
                return
            
            # Fetch unread emails
            messages = await client.call_tool("google_gmail_list_messages", {
                "max_results": 10,
                "query": "is:unread"
            })
            
            if not messages:
                print("No unread emails")
                return
            
            # Build summary
            summary = f"📧 You have {len(messages)} unread emails:\n"
            for msg in messages[:5]:
                subject = msg.get("subject", "No subject")
                sender = msg.get("from", "Unknown")
                summary += f"• {subject} (from {sender})\n"
            
            if len(messages) > 5:
                summary += f"... and {len(messages) - 5} more"
            
            print(summary)
            
            # Post to Slack if available
            if has_slack:
                await client.call_tool("slack_post_message", {
                    "channel": "#daily-digest",
                    "text": summary
                })
                print("Summary posted to Slack!")
                
    except AuthenticationError as e:
        print(f"Authentication failed: {e}")
        print("Run: afctl auth login")
    except MCPError as e:
        print(f"Tool error: {e.message}")

if __name__ == "__main__":
    asyncio.run(email_assistant())copy

Web App with Keycloak

from fastapi import FastAPI, Depends, HTTPException
from af_sdk import MCPClient
from af_sdk.exceptions import AuthenticationError, MCPError

app = FastAPI()

APP_ID = "org-abc123_web-app"
APP_SECRET = "sk_live_xyz789..."

async def get_mcp_client(keycloak_token: str):
    """Create an MCP client for the authenticated user."""
    client = MCPClient(
        method="keycloak",
        app_id=APP_ID,
        app_secret=APP_SECRET,
        keycloak_token=keycloak_token
    )
    await client.connect()
    return client

@app.post("/api/send-email")
async def send_email(
    to: str,
    subject: str,
    body: str,
    token: str = Depends(get_keycloak_token)  # Your auth dependency
):
    try:
        client = await get_mcp_client(token)
        try:
            result = await client.call_tool("google_gmail_send_email", {
                "to": to,
                "subject": subject,
                "body": body
            })
            return {"success": True, "result": result}
        finally:
            await client.disconnect()
            
    except AuthenticationError:
        raise HTTPException(401, "Authentication failed")
    except MCPError as e:
        raise HTTPException(500, f"Tool error: {e.message}")

@app.get("/api/tools")
async def list_tools(token: str = Depends(get_keycloak_token)):
    try:
        client = await get_mcp_client(token)
        try:
            tools = await client.list_tools()
            return {"tools": client.tool_names}
        finally:
            await client.disconnect()
    except AuthenticationError:
        raise HTTPException(401, "Authentication failed")copy

Synchronous Script

from af_sdk import MCPClient
from af_sdk.exceptions import MCPError

def daily_report():
    """Sync script that generates a daily report."""
    
    with MCPClient(
        method="cli",
        app_id="org-abc123_reporter",
        app_secret="sk_live_xyz789..."
    ) as client:
        # List tools (sync)
        client.list_tools_sync()
        
        # Get calendar events
        if client.has_tool("google_calendar_list_events"):
            events = client.call_tool_sync("google_calendar_list_events", {
                "time_min": "2024-01-15T00:00:00Z",
                "time_max": "2024-01-15T23:59:59Z"
            })
            print(f"Today's events: {len(events)}")
        
        # Get unread emails count
        if client.has_tool("google_gmail_list_messages"):
            messages = client.call_tool_sync("google_gmail_list_messages", {
                "max_results": 100,
                "query": "is:unread"
            })
            print(f"Unread emails: {len(messages)}")

if __name__ == "__main__":
    daily_report()copy