Python client (pact-mcp-client)
Call Pact's MCP server from async Python, with LangChain, AutoGen, and LlamaIndex adapters for building agents.
pact-mcp-client is the official async Python client for Pact's MCP server. Same
surface as the TypeScript client, idiomatic Python, plus adapters that drop Pact's
tools straight into the AI agent ecosystem.
Install
pip install pact-mcp-client
# with an ecosystem adapter:
pip install "pact-mcp-client[langchain]" # or [autogen] / [llamaindex]
Quick start
import asyncio
from pact_mcp import PactMCPClient
async def main():
async with PactMCPClient(token="pact_live_...") as pact:
accounts = await pact.query_accounts(search="acme", limit=5)
print(accounts["total"], "accounts;", accounts["consent_filtered"], "hidden by consent")
asyncio.run(main())
Use oauth=OAuthClientCredentials(client_id, client_secret) instead of token
for a tenant-scoped OAuth app. Default endpoint https://api.pact.place/mcp
(override with base_url=).
A Pact-backed agent in under 20 lines
import asyncio, os
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pact_mcp import PactMCPClient
from pact_mcp.integrations import to_langchain_tools
async def main():
async with PactMCPClient(token=os.environ["PACT_API_KEY"]) as pact:
tools = await to_langchain_tools(pact)
prompt = ChatPromptTemplate.from_messages(
[("system", "You are a revenue analyst with live CRM access via Pact."),
("human", "{input}"), ("placeholder", "{agent_scratchpad}")])
agent = create_tool_calling_agent(ChatOpenAI(model="gpt-4o-mini"), tools, prompt)
out = await AgentExecutor(agent=agent, tools=tools).ainvoke(
{"input": "How healthy is our pipeline and which industries lead?"})
print(out["output"], f"\nPact spend: {pact.total_cost_cents:.3f}¢")
asyncio.run(main())
Ecosystem adapters
Each adapter derives a proper args schema from the tool's live JSON schema, so the
LLM sees real parameters. Filter with include= / exclude=.
from pact_mcp.integrations import (
to_langchain_tools, to_autogen_tools, to_llamaindex_tools, pact_tools,
)
lc = await to_langchain_tools(pact) # list[StructuredTool]
ag = await to_autogen_tools(pact) # list[autogen_core.tools.FunctionTool]
li = await to_llamaindex_tools(pact) # list[llama_index.core.tools.FunctionTool]
generic = await pact_tools(pact, include=["query_deals"]) # framework-agnostic PactTool
Typed tool wrappers
await pact.query_accounts(search="acme", limit=10)
await pact.query_contacts(company="Acme")
await pact.query_deals(forecast_cat="Commit")
await pact.query_pipeline_health()
await pact.get_metric_explanation(metric="win_rate", range="30d")
await pact.ask_workspace(question="Where is pipeline slipping?")
await pact.list_agents()
await pact.fire_agent(agent_id="deal_coach_agent", target_type="deal", target_id="123")
await pact.read_briefing()
Consent, cost, errors
pact = PactMCPClient(token=key, on_cost=lambda e: meter(e["tool"], e["cost_cents"]))
res = await pact.fire_agent(agent_id="deal_coach_agent", target_type="contact", target_id="c_1")
if res.get("blocked"):
print(res["reason"]) # "consent_withdrawn"
print(pact.total_cost_cents)
Failures raise a PactMCPError subclass: PactAuthError, PactScopeError,
PactRateLimitError (retried automatically), PactCostCapError, PactToolError.