Skip to main content

Build a Custom MCP Client

Connect to FullEnrich MCP from your own code. No Claude, ChatGPT, or Manus required.

Written by Greg Démogé

Who is this for?

If you're building your own AI agent (using OpenAI Agent SDK, LangChain, CrewAI, or plain Python) and want to call FullEnrich tools directly, this guide is for you.

FullEnrich MCP works with any MCP-compatible client. Claude Desktop and ChatGPT are just two examples; you can build your own.

What you'll build

By the end of this guide, your code will be able to:

  • Authenticate with your FullEnrich account via OAuth

  • List all 11 available MCP tools

  • Call any tool programmatically (search, enrich, export)

  • Persist tokens so you only log in once

Prerequisites

  • Python 3.10+

  • A FullEnrich account with credits

Step 1: Install dependencies

pip install "mcp[cli]" httpx aiohttp

Step 2: Set up OAuth token storage

FullEnrich MCP uses OAuth. On first connection, your browser opens for login. After that, tokens are reused automatically.

Create a file called token_storage.py:

import json
from pathlib import Path
from mcp.shared.auth import OAuthClientInformationFull, OAuthToken

TOKEN_FILE = Path(__file__).parent / ".oauth_tokens.json"


class FileTokenStorage:
"""Persist OAuth tokens to a local JSON file."""

def _load(self) -> dict:
if TOKEN_FILE.exists():
return json.loads(TOKEN_FILE.read_text())
return {}

def _save(self, data: dict):
TOKEN_FILE.write_text(json.dumps(data, indent=2))

async def get_tokens(self) -> OAuthToken | None:
data = self._load()
if "tokens" in data:
return OAuthToken(**data["tokens"])
return None

async def set_tokens(self, tokens: OAuthToken) -> None:
data = self._load()
data["tokens"] = tokens.model_dump(mode="json")
self._save(data)

async def get_client_info(self) -> OAuthClientInformationFull | None:
data = self._load()
if "client_info" in data:
return OAuthClientInformationFull(**data["client_info"])
return None

async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:
data = self._load()
data["client_info"] = client_info.model_dump(mode="json")
self._save(data)

Step 3: Create the MCP client

Create fullenrich_client.py:

import asyncio
import webbrowser
from aiohttp import web
from mcp.client.streamable_http import streamablehttp_client
from mcp.client.auth import OAuthClientProvider
from mcp.shared.auth import OAuthClientMetadata
from mcp import ClientSession
from token_storage import FileTokenStorage

FULLENRICH_MCP_URL = "https://mcp.fullenrich.com/mcp"
CALLBACK_PORT = 9876
REDIRECT_URI = f"http://localhost:{CALLBACK_PORT}/callback"

_auth_response = None


async def start_callback_server():
global _auth_response
_auth_response = None

async def handle_callback(request):
global _auth_response
_auth_response = dict(request.query)
return web.Response(
text="<h2>Authentication successful!</h2><p>You can close this tab.</p>",
content_type="text/html",
)

app = web.Application()
app.router.add_get("/callback", handle_callback)
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, "localhost", CALLBACK_PORT)
await site.start()
return runner


async def redirect_handler(auth_url: str) -> None:
print("Opening browser for FullEnrich login...")
webbrowser.open(auth_url)


async def callback_handler() -> tuple[str, str | None]:
global _auth_response
print("Waiting for login to complete...")
while _auth_response is None:
await asyncio.sleep(0.5)
return _auth_response.get("code", ""), _auth_response.get("state")


async def main():
storage = FileTokenStorage()

existing = await storage.get_tokens()
if existing:
print("Reusing saved OAuth tokens.")
else:
print("First connection - browser will open for login.")

oauth_provider = OAuthClientProvider(
server_url=FULLENRICH_MCP_URL,
client_metadata=OAuthClientMetadata(
redirect_uris=[REDIRECT_URI],
token_endpoint_auth_method="none",
grant_types=["authorization_code", "refresh_token"],
response_types=["code"],
client_name="My Custom MCP Client",
),
storage=storage,
redirect_handler=redirect_handler,
callback_handler=callback_handler,
timeout=120.0,
)

callback_runner = await start_callback_server()

try:
async with streamablehttp_client(
url=FULLENRICH_MCP_URL,
auth=oauth_provider,
) as (read_stream, write_stream, _):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
print("Connected to FullEnrich MCP!\n")

result = await session.call_tool("get_credits", arguments={})
print(f"Credits: {result.content[0].text}")

result = await session.call_tool("search_people", arguments={
"current_position_titles": [
{"value": "Head of Sales", "exact_match": False}
],
"current_company_headcounts": ["51-200"],
"limit": 5,
})
print(f"\nSearch results:\n{result.content[0].text}")

finally:
await callback_runner.cleanup()


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

Step 4: Run it

python fullenrich_client.py

On first run, your browser opens and you log in with your FullEnrich account. Tokens are saved to .oauth_tokens.json. Every subsequent run reuses them automatically, no browser needed.

Available tools

Tool

Description

Credits

get_credits

Check your credit balance

Free

list_industries

List valid industry filter values

Free

search_people

Search contacts with filters

Free

search_companies

Search companies with filters

Free

enrich_search_contact

Enrich contacts from search filters

Paid

enrich_bulk

Enrich a list of known contacts

Paid

search_contact_by_email

Reverse email lookup

Paid

get_enrichment_results

Poll enrichment job status

Free

export_contacts

Export contact results to CSV/JSON

Free

export_companies

Export company results to CSV/JSON

Free

export_enrichment_results

Export enrichment results to file

Free

Plug into your AI agent framework

Once you can call tools, you can bridge them into any agent framework.

OpenAI Agent SDK

from agents import Agent
from agents.mcp import MCPServerStreamableHttp

agent = Agent(
name="Prospecting Agent",
instructions="You help find and enrich B2B contacts using FullEnrich.",
mcp_servers=[
MCPServerStreamableHttp(
url="https://mcp.fullenrich.com/mcp",
# Pass your OAuth headers here
)
],
)

Any framework (generic bridge)

If your framework doesn't support MCP natively, use the pattern from Step 3: connect with the MCP SDK, then wrap session.call_tool() calls in your framework's tool format.

Adding skills for better agent behavior

FullEnrich publishes workflow skills, markdown files that describe best practices for using the tools (confirmation before enrichment, credit checking, filter strategies).

These aren't code. They're instructions you can inject into your agent's system prompt:

from agents import Agent
from agents.mcp import MCPServerStreamableHttp

agent = Agent(
name="Prospecting Agent",
instructions="You help find and enrich B2B contacts using FullEnrich.",
mcp_servers=[
MCPServerStreamableHttp(
url="https://mcp.fullenrich.com/mcp",
# Pass your OAuth headers here
)
],
)

This gives your custom agent the same guided behavior that Claude users get with FullEnrich skills.

Security notes

  • .oauth_tokens.json contains your access tokens. Treat it like a password and add it to .gitignore.

  • Enrichment costs credits. Always call get_credits first and confirm with the user before running paid operations.

Troubleshooting

Issue

Fix

401 Unauthorized

Delete .oauth_tokens.json and re-authenticate

Browser opens but nothing happens

Make sure port 9876 is free, or change CALLBACK_PORT

unknown tool error

Call list_tools() to see exact tool names (they may change)

Token refresh fails

Delete .oauth_tokens.json and re-authenticate

Did this answer your question?