A Model Context Protocol server that gives Claude a read-only window into your Outreach org: sequence performance, what is stalled mid-sequence, prospect search, and one prospect’s engagement history. Your SDR manager asks “what is paused in the Q3 enterprise sequence and why” in chat and gets rows with pause reasons attached, from a process that has no code path capable of changing anything. The scaffold lives at apps/web/public/artifacts/mcp-server-outreach-revops/ — a README.md, a pyproject.toml, and src/outreach_revops_mcp/server.py, installable with pip install -e ..
Read the next section before you build it, because Outreach already ships one.
When to use this
Outreach hosts its own MCP server at https://api.outreach.io/mcp/. It authenticates over OAuth 2.1 with user-level identity, follows the MCP authorization standard published 2025-11-11, and exposes tools in six categories — workflow, prospecting, account, deal, user, and calendar. It requires the Amplify add-on enabled on the seat plus an admin toggle in Organization settings, and it is read, create, and delete only: Outreach deliberately excludes record updates, on the reasoning that model behavior when editing existing records is unpredictable (vendor docs, Outreach support portal).
For most teams the hosted server is the right answer and this scaffold is wasted work. Turn it on, connect it, move on. Build your own when one of four things is true.
The agent should not be able to delete a prospect. The hosted server’s prospecting category includes create and delete. Delete is the one Outreach operation with no undo and no local copy — a deleted prospect takes its sequence history with it. The scaffold has no POST, PATCH, or DELETE anywhere in its dispatch table, so an instruction that reaches the model through a prospect’s own notes field has nothing to call. That is a structural property, not a policy someone has to enforce.
You need a service-account identity. The hosted server runs as the signed-in human with that human’s permissions. An agent wired into a Slack channel, a nightly reporting job, or a workflow the whole team triggers has no individual human behind it, and a per-user OAuth grant cannot express “less than any one person sees.”
Amplify is not on every seat. The hosted server is gated on the add-on. Third-party pricing research puts 2026 Amplify tiers at roughly $100, $130, and $160 per user per month for Core, Plus, and Pro — Outreach does not publish these, so treat them as reported bands, not quotes. A standard OAuth application against the public API has no such gate, so a 40-seat org can answer sequence questions in chat without buying Amplify for 40 people to do it.
You want aggregate reads.get_sequence_performance answers “how is this sequence doing” in a single request against counters Outreach maintains itself.
When NOT to use this
You have no reason to reject the hosted server. Repeating it because it is the most common mistake here: the default is Outreach’s own server, and four narrow cases are the whole argument for anything else.
Prospect data cannot reach an LLM. Every row returned carries names, work emails, titles, and engagement history into the conversation. OUTREACH_ALLOWED_SEQUENCE_IDS narrows the surface; it does not eliminate it. If your policy forbids contact data in a third-party model, neither server is the right project.
You want the agent to run sequences. Adding prospects to sequences, pausing them, sending mailings — none of that is here, by design. Use the hosted server, which does create, or the Outreach UI.
The question is a bulk export. Every tool caps at 100 rows and returns one page. A quarterly rollup across every sequence is a script against /api/v2/sequences with pagination, reviewed as a file. Chat is the wrong interface for 4,000 rows.
What it exposes
Five tools, all reads.
list_sequences hits GET /sequences sorted by -lastUsedAt, returning each sequence’s engagement counters. This is the id-lookup step before anything else.
get_sequence_performance hits GET /sequences/{id} and adds a derived block: prospect reply rate, bounce rate, opt-out rate, with the raw counters under _basis so a human can check the arithmetic against the Outreach UI.
find_stalled_sequence_states hits GET /sequenceStates filtered on state, including prospect and sequence, sorted by -stateChangedAt. It carries pauseReason and errorReason through so “what is stuck” comes back with the reason rather than a count.
search_prospects hits GET /prospects with a fixed 15-field projection and computes a contactable_count that excludes opted-out records.
get_prospect_engagement hits GET /prospects/{id} plus GET /mailings filtered to that prospect, giving delivery, open, click, reply, and bounce timestamps for the last ten sends.
Engineering posture
Three choices in server.py are load-bearing.
Every request carries an explicit sparse fieldset. The Outreach prospect resource defines 230 attributes, 150 of which are custom1 through custom150 (verified against the org’s OpenAPI definition at https://api.outreach.io/api/v2/schema/openapi.json). The default response is mostly nulls, and you pay tokens for all of them on every row. PROSPECT_FIELDS projects to 15. The custom fields are excluded deliberately: those slots are where orgs park comp bands, contract terms, and notes nobody intended to publish, and a field named custom17 gives the model no way to know what it is reading.
Filter keys are checked before the request goes out. Outreach marks a subset of each resource’s attributes filterable — 17 of the prospect’s 230. An unsupported filter is not rejected upstream. The parameter is ignored, a 200 comes back with the full collection, and the model reports the org-wide count as if it were the filtered answer. _check_filters() refuses any key outside the verified set and returns the allowed list plus a note on the common misses: prospect company, optedOut, and emailOptedOut are all returned but none of them is filterable. search_prospects therefore computes contactable_count client-side rather than pretending a filter exists.
Reply rate is computed per prospect, not per message._rates() divides numRepliedProspects by numContactedProspects rather than replyCount by deliverCount. replyCount counts messages, so one engaged prospect replying four times reads as four replies against four separate sends and inflates the rate on exactly the sequences a manager is trying to evaluate.
Failure modes and guards
The rotated refresh token gets lost, and auth dies two hours later. Outreach access tokens last 2 hours; each refresh issues a new refresh token and retires the one used. A server that holds the new token only in memory works until it restarts, then presents a dead credential — surfacing as a 401 that reads like a scope problem. Guard:TokenStore._refresh() writes the rotated token to OUTREACH_TOKEN_FILE via a temp-file rename before the new access token is returned to any caller, and TokenStore.load() test-writes that file at startup and refuses to run if it is not writable. Refresh tokens also expire 14 days after issue, so a server idle longer than that needs the authorization code flow re-run; the error message says so explicitly.
An agent loop drains the org’s API budget. Outreach allows 10,000 requests per hour per user and returns X-RateLimit-Limit, X-RateLimit-Remaining, and X-RateLimit-Reset on every response (vendor docs). That budget is shared with your CRM sync and every other integration on the org, so an agent that pages hard breaks the Salesforce sync, not just chat. Guard:_get() reads X-RateLimit-Remaining on every response and raises once it drops below OUTREACH_RATE_LIMIT_FLOOR, default 250, naming the reset time. Set it higher — 500 or more — on an org where the sync matters.
Included resources smuggle in the payload the projection just removed.find_stalled_sequence_states uses include=prospect,sequence, and JSON:API returns included resources at full width unless they are projected too. Fifty stalled rows each drag a 230-attribute prospect behind them. Guard: the extra_fields argument sets fields[prospect] and fields[sequence] alongside fields[sequenceState], holding included prospects to five attributes.
A truncated answer reads as a complete one. Every tool caps at page[limit]=100 and returns the first page only. Guard: partial — the cap is enforced and documented, but the tools do not yet flag truncation. It is item 2 on the numbered pre-production list in README.md, and it is the first thing to fix if anyone starts quoting these numbers upward.
Instead of building this
Beyond the hosted server, CData publishes a read-only Outreach MCP server built on its JDBC driver, and Zapier and Pipedream both expose Outreach through their generic MCP layers. All three are faster to stand up than this scaffold. The reason to pass on them is the same reason to pass on the hosted server: the credential and the data path belong to a third party. This scaffold’s tool surface, its scope set, and its rate-limit floor are values in a file you own — which matters when the answer to “what could that agent see” has to be an inspection rather than a vendor claim.
If you are building the same read-mostly posture across systems of record, the Apollo and Gong servers in this series share the projection-and-preflight shape, so prompts stay portable between them. For the difference between shipping this as a server versus a packaged skill, see Claude Skill vs MCP server.
# outreach-revops-mcp
A read-only MCP server over the [Outreach REST API v2](https://developers.outreach.io/api/). It gives Claude five tools — sequence listing, sequence performance, stalled sequence states, prospect search, and single-prospect engagement — and no way to write anything.
**Read this first: Outreach ships its own MCP server.** It runs at `https://api.outreach.io/mcp/`, authenticates the signed-in user over OAuth 2.1, and exposes read, create, and delete tools across workflow, prospecting, account, deal, user, and calendar categories. It requires the Amplify add-on on the seat and an admin toggle in Organization settings. If that fits, install it and stop reading — this scaffold is wasted work.
Build this instead when one of these is true:
- **The agent should not be able to delete a prospect.** The hosted server exposes prospect create and delete. This one has no `POST`, `PATCH`, or `DELETE` in its dispatch table, so a prompt-injected instruction to "clean up these duplicates" has nothing to call.
- **You need a service-account identity.** The hosted server runs as the signed-in human with that human's permissions. A shared agent — one wired into Slack, a reporting job, something the whole team triggers — cannot be expressed that way.
- **Not every seat has Amplify.** The hosted server is gated on the add-on per user. A standard API OAuth application is not.
- **You want aggregate reads.** `get_sequence_performance` answers "how is this sequence doing" in one request against Outreach's own pre-aggregated counters instead of paging sequence states.
**Status: scaffold, not runtime-tested.** Endpoint paths, attribute names, filterable-attribute sets, and query syntax were transcribed from the machine-readable OpenAPI definition at `https://api.outreach.io/api/v2/schema/openapi.json` and the developer portal as of 2026-08. Custom fields are per-org and are not in that definition. Verify against your own org before relying on it.
## Install
```bash
cd mcp-server-outreach-revops
pip install -e .
```
Requires Python 3.11+.
## Environment variables
### `OUTREACH_CLIENT_ID` / `OUTREACH_CLIENT_SECRET` (required)
Register an application at [developers.outreach.io](https://developers.outreach.io/) under your Outreach org. The identifier and secret appear on the application page after creation. Request exactly these scopes — the server needs no others and asking for more widens the blast radius of a leaked token:
```
prospects.read
sequences.read
sequenceStates.read
mailings.read
```
Outreach scopes are `<pluralized-resource>.<read|write|delete|all>`. Do not request `.all` on anything.
### `OUTREACH_REDIRECT_URI` (required)
The exact redirect URI registered on the application. It is sent again on every refresh, and a mismatch fails the refresh with a 400 that reads like a credential problem.
### `OUTREACH_TOKEN_FILE` (default `~/.outreach-mcp-token.json`)
Where the rotating refresh token lives. Complete the authorization code flow once by hand, then write the result:
```bash
echo '{"refresh_token":"PASTE_REFRESH_TOKEN_HERE"}' > ~/.outreach-mcp-token.json
chmod 600 ~/.outreach-mcp-token.json
```
**This file is the grant.** Outreach issues a new refresh token with every access token and retires the old one. The server writes the new value before using the new access token, and refuses to start if the file is not writable — a read-only token file produces a server that works for two hours and then 401s on everything.
### `OUTREACH_ALLOWED_SEQUENCE_IDS` (optional, comma-separated)
Numeric sequence ids the agent may read. Empty means no restriction. Set it when sequence names carry customer or campaign names that should not reach an LLM, or when a shared agent should only see its own team's sequences.
### `OUTREACH_RATE_LIMIT_FLOOR` (default `250`)
The server stops answering when fewer than this many of the org's 10,000 hourly API calls remain. That budget is shared with your CRM sync and every other integration on the org, so an agent loop that drains it breaks more than chat.
### `OUTREACH_BASE_URL` / `OUTREACH_TOKEN_URL` (optional)
Default to `https://api.outreach.io/api/v2` and `https://api.outreach.io/oauth/token`.
## Register with Claude
Claude Desktop — `claude_desktop_config.json`:
```json
{
"mcpServers": {
"outreach-revops": {
"command": "python",
"args": ["-m", "outreach_revops_mcp.server"],
"env": {
"OUTREACH_CLIENT_ID": "...",
"OUTREACH_CLIENT_SECRET": "...",
"OUTREACH_REDIRECT_URI": "https://example.com/oauth/callback",
"OUTREACH_TOKEN_FILE": "/Users/you/.outreach-mcp-token.json",
"OUTREACH_RATE_LIMIT_FLOOR": "500"
}
}
}
}
```
Claude Code:
```bash
claude mcp add outreach-revops -- python -m outreach_revops_mcp.server
```
## Sanity check
Ask, in order:
1. **"List my 5 most recently used Outreach sequences."** — exercises `list_sequences`, the token refresh, and the sparse fieldset. If this 401s, the refresh token is stale or the redirect URI does not match.
2. **"How is sequence 1234 performing?"** — exercises `get_sequence_performance`. The `derived` block should show `prospect_reply_rate_pct` computed from `numRepliedProspects / numContactedProspects`, with the raw counters under `_basis` so you can check the arithmetic against the Outreach UI.
3. **"Find prospects at companies called Acme."** — should fail. `company` is not a filterable prospect attribute, and the server refuses rather than returning an unfiltered list. The error names the filterable set. This is the check that the preflight is working; if it returns rows, `_check_filters` is not being reached.
4. **"What is paused in sequence 1234?"** — exercises `find_stalled_sequence_states` and confirms included prospects come back projected to five fields rather than 230.
## Security model
The OAuth application's token carries four read scopes and nothing else. Anything the tools return enters the conversation: prospect names, work emails, job titles, engagement history, sequence names. `OUTREACH_ALLOWED_SEQUENCE_IDS` narrows that; it does not eliminate it. If prospect data cannot reach an LLM at all under your policy, do not run this or the hosted server.
The token file is the sensitive artifact — it holds a credential that regenerates access indefinitely until it expires or an admin revokes the application. Keep it at mode 600, outside any directory the agent can read as a file, and outside version control.
Revocation is per-application in Outreach admin settings, which kills every token issued to it at once.
## Known limits — do these before production
1. **No test suite.** `pyproject.toml` declares `pytest` and `pytest-httpx` under `dev` but ships no tests. Write them against recorded fixtures before anyone trusts a number out of `_rates()`.
2. **No pagination.** Every tool caps at `page[limit]=100` and returns the first page. A question whose honest answer needs 400 rows silently gets 100. Add cursor following, or have the tools report when a result is truncated.
3. **`FILTERABLE` is a transcription and will drift.** It was copied from the OpenAPI definition's filterable badges. When Outreach adds a filterable attribute, this server keeps rejecting it. Regenerate the sets from `https://api.outreach.io/api/v2/schema/openapi.json` on a schedule rather than by hand.
4. **Custom fields are invisible.** `custom1`–`custom150` on prospects and opportunities are excluded from the projections deliberately. If your org keeps something load-bearing in `custom17`, add it to `PROSPECT_FIELDS` and know what it holds first — these fields are where orgs put comp bands, contract terms, and notes nobody meant to publish.
5. **No audit log.** Tool calls go nowhere. Add structured logging of `(timestamp, tool, arguments, row count)` if you need to answer "what did the agent look at" later.
6. **Rate-limit accounting is per-response, not global.** The floor check reads `X-RateLimit-Remaining` off each response, so a burst of concurrent calls can overshoot before any of them sees a low number.
7. **`_rates()` divides by Outreach's counters, not yours.** `numContactedProspects` counts prospects the sequence contacted, which is not the same denominator your reporting layer may use. Reconcile once against a sequence you know before quoting the output to a leadership audience.
## Files
```
mcp-server-outreach-revops/
├── README.md
├── pyproject.toml
└── src/outreach_revops_mcp/
├── __init__.py
└── server.py
```
`server.py` holds the configuration block, the `TokenStore` refresh-rotation logic, the `FILTERABLE` sets and `_check_filters` preflight, the sparse-fieldset constants, the five tool definitions, and their handlers.
"""
outreach-revops-mcp — a read-only MCP server over the Outreach REST API v2.
Five tools: sequence listing, sequence performance, stalled/errored sequence states,
prospect search, and single-prospect engagement. No POST, PATCH, or DELETE path
exists anywhere in the dispatch table, so no instruction reaching the model can
write to or delete from Outreach through this process.
This exists alongside Outreach's own hosted MCP server at https://api.outreach.io/mcp/.
That server authenticates the individual user over OAuth 2.1, requires the Amplify
add-on on the seat, and exposes read, create, and delete tools across prospecting,
accounts, deals, users, and calendar. Use it when you want breadth and per-user
identity. Use this scaffold when you want a service-account identity, a surface with
no create or delete on it, and aggregate reads that do not page the whole org.
Three engineering choices are load-bearing and documented at their call sites:
1. Every request carries an explicit sparse fieldset. The prospect resource
defines 230 attributes, 150 of which are custom1..custom150; the default
payload is mostly nulls that cost tokens.
2. Filter keys are checked against the attributes Outreach actually marks
filterable before the request goes out. An unrecognized filter is not an
error upstream — it comes back as an unfiltered list, which the model then
reports as a real answer.
3. Refresh tokens rotate on every use and are persisted before the new access
token is returned. Losing the rotated token ends the grant.
STATUS: scaffold — not runtime-tested against a live Outreach org. Endpoint paths,
attribute names, filterable-attribute sets, and query syntax track the machine-
readable OpenAPI definition at https://api.outreach.io/api/v2/schema/openapi.json
and the developer portal (developers.outreach.io) as of 2026-08. Verify against your
own org before relying on it; custom fields are per-org and not in that definition.
Run as: python -m outreach_revops_mcp.server
"""
from __future__ import annotations
import asyncio
import json
import os
import time
from pathlib import Path
from typing import Any
import httpx
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import TextContent, Tool
# ----- Configuration (read from env at startup) -----
OUTREACH_BASE_URL = os.environ.get("OUTREACH_BASE_URL", "https://api.outreach.io/api/v2").rstrip("/")
OUTREACH_TOKEN_URL = os.environ.get("OUTREACH_TOKEN_URL", "https://api.outreach.io/oauth/token")
OUTREACH_CLIENT_ID = os.environ.get("OUTREACH_CLIENT_ID")
OUTREACH_CLIENT_SECRET = os.environ.get("OUTREACH_CLIENT_SECRET")
OUTREACH_REDIRECT_URI = os.environ.get("OUTREACH_REDIRECT_URI")
# Where the rotated refresh token lives. Outreach issues a NEW refresh token with
# every access token and invalidates the old one, so this file is the grant. If it
# is not writable the server refuses to start rather than dying silently in 2 hours.
OUTREACH_TOKEN_FILE = Path(os.environ.get("OUTREACH_TOKEN_FILE", "~/.outreach-mcp-token.json")).expanduser()
# Sequences the agent may read at all, by numeric id. Empty means no restriction.
# Populate this when some sequences carry customer names in their titles, or when a
# shared agent should only see the sequences its own team runs.
OUTREACH_ALLOWED_SEQUENCE_IDS = {
s.strip() for s in os.environ.get("OUTREACH_ALLOWED_SEQUENCE_IDS", "").split(",") if s.strip()
}
# Outreach allows 10,000 requests per hour per user and returns the remaining count
# on every response. Below this floor the server stops answering rather than burning
# the last of the org's budget on a chatty agent loop.
RATE_LIMIT_FLOOR = int(os.environ.get("OUTREACH_RATE_LIMIT_FLOOR", "250"))
MAX_LIMIT = 100
DEFAULT_LIMIT = 25
# ----- Sparse fieldsets -----
#
# Once fields[<type>] is supplied, Outreach returns only the attributes named, so
# each list below must be complete for its tool. These are deliberately short: the
# cost of an over-wide projection is paid on every row of every answer.
PROSPECT_FIELDS = [
"firstName", "lastName", "title", "company", "occupation",
"emails", "optedOut", "emailOptedOut", "callOptedOut",
"engagedScore", "engagedAt", "touchedAt",
"openCount", "clickCount", "replyCount",
]
SEQUENCE_FIELDS = [
"name", "enabled", "locked", "shareType", "sequenceType", "salesMotion",
"sequenceStepCount", "durationInDays", "lastUsedAt",
"numContactedProspects", "numRepliedProspects",
"deliverCount", "openCount", "clickCount", "replyCount",
"bounceCount", "optOutCount", "failureCount", "scheduleCount",
"positiveReplyCount", "negativeReplyCount", "neutralReplyCount",
"throttleMaxAddsPerDay", "throttlePaused",
]
SEQUENCE_STATE_FIELDS = [
"state", "stateChangedAt", "activeAt", "pauseReason", "errorReason",
"deliverCount", "openCount", "clickCount", "replyCount",
"bounceCount", "failureCount", "optOutCount", "repliedAt", "callCompletedAt",
]
MAILING_FIELDS = [
"subject", "mailingType", "state", "stateChangedAt",
"scheduledAt", "deliveredAt", "openedAt", "clickedAt", "repliedAt",
"bouncedAt", "unsubscribedAt", "markedAsSpamAt", "errorReason",
]
# ----- Filterable attributes -----
#
# Outreach marks a subset of each resource's attributes as filterable. A filter on
# anything else does not 400 — the parameter is ignored and the full collection
# comes back. The model then answers "3,812 prospects match" for a filter that never
# applied. These sets are transcribed from the OpenAPI definition's filterable
# badges and are the preflight check in _check_filters().
FILTERABLE: dict[str, set[str]] = {
"prospect": {
"createdAt", "updatedAt", "emails", "engagedAt", "engagedScore",
"externalSource", "firstName", "lastName", "githubUsername",
"linkedInId", "linkedInSlug", "sharingTeamId", "stackOverflowId",
"timeZone", "title", "touchedAt", "twitterUsername",
},
"sequence": {
"createdAt", "updatedAt", "name", "clickCount", "deliverCount",
"enabledAt", "lastUsedAt", "lockedAt", "openCount", "replyCount",
"salesMotion", "shareType", "throttleCapacity", "throttleMaxAddsPerDay",
},
"sequenceState": {
"createdAt", "updatedAt", "state", "stateChangedAt", "pauseReason",
"callCompletedAt", "clickCount", "deliverCount", "openCount",
"repliedAt", "replyCount",
},
"mailing": {
"createdAt", "updatedAt", "state", "stateChangedAt", "mailingType",
"messageId", "bouncedAt", "clickedAt", "deliveredAt", "openedAt",
"repliedAt", "retryAt", "scheduledAt", "unsubscribedAt",
"notifyThreadScheduledAt", "notifyThreadStatus",
},
}
# Relationship filters are addressed as filter[<relationship>][id] and are not part
# of the attribute badge set above, so they get their own allowlist per resource.
FILTERABLE_RELATIONSHIPS: dict[str, set[str]] = {
"prospect": {"account", "owner", "stage"},
"sequence": {"owner", "creator"},
"sequenceState": {"prospect", "sequence", "mailbox", "user", "account"},
"mailing": {"prospect", "sequence", "mailbox", "user"},
}
# Attributes readers most often want to filter on that Outreach does not support as
# filters. Naming them in the error is the difference between the agent adapting and
# the agent inventing a workaround.
KNOWN_UNFILTERABLE_HINT = {
"prospect": "company, optedOut, emailOptedOut, callOptedOut, tags, and openCount "
"are returned but not filterable — fetch and filter client-side",
"sequence": "enabled, locked, and sequenceType are returned but not filterable",
"sequenceState": "errorReason and activeAt are returned but not filterable",
"mailing": "subject is returned but not filterable",
}
class OutreachError(RuntimeError):
"""Raised for configuration, auth, and upstream failures surfaced to the model."""
# ----- Token handling -----
class TokenStore:
"""
Holds the access token and persists the rotating refresh token.
Outreach access tokens live 2 hours. Each refresh returns a new refresh token and
retires the one used; the old value is dead the moment the new one is issued. The
write therefore happens BEFORE the new access token is handed to a caller — if the
process dies between the two, a saved-but-unused refresh token still works, while
an unsaved one loses the grant and forces a manual re-authorization.
"""
def __init__(self, path: Path) -> None:
self.path = path
self._access_token: str | None = None
self._expires_at: float = 0.0
self._refresh_token: str | None = None
self._lock = asyncio.Lock()
def load(self) -> None:
if not self.path.exists():
raise OutreachError(
f"token file {self.path} not found — complete the OAuth authorization code "
f"flow once and write {{'refresh_token': '...'}} to it (see README)"
)
data = json.loads(self.path.read_text(encoding="utf-8"))
self._refresh_token = data.get("refresh_token")
if not self._refresh_token:
raise OutreachError(f"token file {self.path} has no 'refresh_token' key")
# Startup writability check. A read-only token file is a server that works for
# 2 hours and then fails every call with a 401 that looks like a scope problem.
try:
self.path.write_text(json.dumps({"refresh_token": self._refresh_token}), encoding="utf-8")
except OSError as exc:
raise OutreachError(f"token file {self.path} is not writable: {exc}") from exc
def _persist(self, refresh_token: str) -> None:
tmp = self.path.with_suffix(self.path.suffix + ".tmp")
tmp.write_text(json.dumps({"refresh_token": refresh_token}), encoding="utf-8")
tmp.replace(self.path)
try:
self.path.chmod(0o600)
except OSError:
pass # Windows and some mounts do not honour chmod; not fatal.
async def access_token(self, client: httpx.AsyncClient) -> str:
async with self._lock:
# 120s of slack so a request issued just under the wire does not land expired.
if self._access_token and time.time() < self._expires_at - 120:
return self._access_token
await self._refresh(client)
assert self._access_token is not None
return self._access_token
async def _refresh(self, client: httpx.AsyncClient) -> None:
if not (OUTREACH_CLIENT_ID and OUTREACH_CLIENT_SECRET and OUTREACH_REDIRECT_URI):
raise OutreachError(
"OUTREACH_CLIENT_ID, OUTREACH_CLIENT_SECRET and OUTREACH_REDIRECT_URI are required"
)
resp = await client.post(
OUTREACH_TOKEN_URL,
data={
"client_id": OUTREACH_CLIENT_ID,
"client_secret": OUTREACH_CLIENT_SECRET,
"redirect_uri": OUTREACH_REDIRECT_URI,
"grant_type": "refresh_token",
"refresh_token": self._refresh_token,
},
)
if resp.status_code != 200:
raise OutreachError(
f"token refresh failed ({resp.status_code}). Refresh tokens expire 14 days "
f"after issue — if this server sat idle longer than that, re-run the "
f"authorization code flow. Body: {resp.text[:300]}"
)
payload = resp.json()
new_refresh = payload.get("refresh_token")
if not new_refresh:
raise OutreachError("token refresh returned no refresh_token; refusing to continue")
self._persist(new_refresh) # persist before use, see class docstring
self._refresh_token = new_refresh
self._access_token = payload["access_token"]
self._expires_at = time.time() + int(payload.get("expires_in", 7200))
TOKENS = TokenStore(OUTREACH_TOKEN_FILE)
# ----- HTTP -----
def _check_filters(resource: str, filters: dict[str, Any] | None) -> dict[str, str]:
"""
Reject filter keys Outreach does not honour, before the request is sent.
Silent-ignore is the failure this guards. Outreach answers a request carrying an
unsupported filter with the unfiltered collection and a 200, so nothing downstream
can tell a narrow answer from a whole-org answer.
"""
if not filters:
return {}
allowed = FILTERABLE.get(resource, set())
allowed_rel = FILTERABLE_RELATIONSHIPS.get(resource, set())
out: dict[str, str] = {}
for key, value in filters.items():
if key in allowed_rel:
out[f"filter[{key}][id]"] = str(value)
elif key in allowed:
out[f"filter[{key}]"] = str(value)
else:
hint = KNOWN_UNFILTERABLE_HINT.get(resource, "")
raise OutreachError(
f"'{key}' is not a filterable {resource} attribute. Outreach ignores unknown "
f"filters and returns everything, so this server refuses the call instead. "
f"Filterable: {', '.join(sorted(allowed | allowed_rel))}."
+ (f" Note: {hint}." if hint else "")
)
return out
async def _get(
client: httpx.AsyncClient,
path: str,
resource: str,
*,
fields: list[str] | None = None,
filters: dict[str, Any] | None = None,
include: str | None = None,
sort: str | None = None,
limit: int | None = None,
extra_fields: dict[str, list[str]] | None = None,
) -> dict[str, Any]:
params: dict[str, str] = {}
params.update(_check_filters(resource, filters))
if fields:
params[f"fields[{resource}]"] = ",".join(fields)
for extra_resource, extra in (extra_fields or {}).items():
params[f"fields[{extra_resource}]"] = ",".join(extra)
if include:
params["include"] = include
if sort:
params["sort"] = sort
if limit is not None:
params["page[limit]"] = str(max(1, min(limit, MAX_LIMIT)))
token = await TOKENS.access_token(client)
resp = await client.get(
f"{OUTREACH_BASE_URL}{path}",
params=params,
headers={
"Authorization": f"Bearer {token}",
"Content-Type": "application/vnd.api+json",
},
)
remaining = resp.headers.get("X-RateLimit-Remaining")
if remaining is not None and remaining.isdigit() and int(remaining) < RATE_LIMIT_FLOOR:
reset = resp.headers.get("X-RateLimit-Reset", "unknown")
raise OutreachError(
f"stopping: {remaining} of the org's 10,000 hourly API calls remain, below the "
f"configured floor of {RATE_LIMIT_FLOOR}. Window resets at {reset}. Outreach's "
f"limit is shared with your CRM sync, so burning it here breaks that too."
)
if resp.status_code == 429:
raise OutreachError("Outreach returned 429. The hourly request budget is exhausted.")
if resp.status_code == 403:
raise OutreachError(
f"403 from Outreach on {path}. The OAuth application is missing a scope — "
f"this server needs prospects.read, sequences.read, sequenceStates.read, "
f"and mailings.read. Body: {resp.text[:200]}"
)
if resp.status_code >= 400:
raise OutreachError(f"{resp.status_code} from Outreach on {path}: {resp.text[:300]}")
return resp.json()
# ----- Shaping -----
def _flatten(item: dict[str, Any]) -> dict[str, Any]:
"""Collapse a JSON:API resource object into {id, ...attributes} with nulls dropped."""
out: dict[str, Any] = {"id": item.get("id")}
for key, value in (item.get("attributes") or {}).items():
if value not in (None, [], ""):
out[key] = value
return out
def _rates(attrs: dict[str, Any]) -> dict[str, Any]:
"""
Derive the rates a human actually asks for from Outreach's raw counters.
Reply rate is computed against numRepliedProspects / numContactedProspects rather
than replyCount / deliverCount: replyCount counts messages, so one prospect replying
four times reads as four replies against four different sends. The prospect-level
pair is the one that answers "is this sequence working".
"""
contacted = attrs.get("numContactedProspects") or 0
replied = attrs.get("numRepliedProspects") or 0
delivered = attrs.get("deliverCount") or 0
bounced = attrs.get("bounceCount") or 0
opted_out = attrs.get("optOutCount") or 0
attempted = delivered + bounced
def pct(num: int, den: int) -> float | None:
return round(100.0 * num / den, 2) if den else None
return {
"prospect_reply_rate_pct": pct(replied, contacted),
"bounce_rate_pct": pct(bounced, attempted),
"opt_out_rate_pct": pct(opted_out, delivered),
"_basis": {
"contacted_prospects": contacted,
"replied_prospects": replied,
"delivered": delivered,
"bounced": bounced,
"opted_out": opted_out,
},
}
def _sequence_allowed(seq_id: str | None) -> bool:
return not OUTREACH_ALLOWED_SEQUENCE_IDS or str(seq_id) in OUTREACH_ALLOWED_SEQUENCE_IDS
def _ok(payload: Any) -> list[TextContent]:
return [TextContent(type="text", text=json.dumps(payload, indent=2, default=str))]
# ----- Tools -----
TOOLS = [
Tool(
name="list_sequences",
description=(
"List Outreach sequences with their engagement counters, newest-used first. "
"Use this to find a sequence id before asking about its performance. Filterable "
"keys: name, salesMotion, shareType, lastUsedAt, createdAt, updatedAt, owner, creator."
),
inputSchema={
"type": "object",
"properties": {
"filters": {
"type": "object",
"description": "Filter keys checked against Outreach's filterable set before sending.",
"additionalProperties": {"type": "string"},
},
"limit": {"type": "integer", "minimum": 1, "maximum": MAX_LIMIT},
},
},
),
Tool(
name="get_sequence_performance",
description=(
"Fetch one sequence and return its counters plus derived prospect reply rate, "
"bounce rate, and opt-out rate. Answers 'how is this sequence doing' in a single "
"API call — do not page sequence states to compute this."
),
inputSchema={
"type": "object",
"properties": {"sequence_id": {"type": "string"}},
"required": ["sequence_id"],
},
),
Tool(
name="find_stalled_sequence_states",
description=(
"Find prospects sitting in a non-running sequence state (paused, bounced, failed, "
"finished) with the pause and error reasons attached. Use for 'what is stuck in "
"sequence X'. State is a filterable attribute; errorReason is not."
),
inputSchema={
"type": "object",
"properties": {
"state": {
"type": "string",
"description": "Outreach sequence state, e.g. paused, bounced, failed, finished, active.",
},
"sequence_id": {"type": "string"},
"limit": {"type": "integer", "minimum": 1, "maximum": MAX_LIMIT},
},
"required": ["state"],
},
),
Tool(
name="search_prospects",
description=(
"Search prospects on filterable attributes and return a fixed 15-field projection "
"including opt-out status. Filterable keys: firstName, lastName, title, emails, "
"engagedScore, engagedAt, touchedAt, createdAt, updatedAt, account, owner, stage. "
"Company name and opt-out flags are NOT filterable — filter those from the results."
),
inputSchema={
"type": "object",
"properties": {
"filters": {"type": "object", "additionalProperties": {"type": "string"}},
"limit": {"type": "integer", "minimum": 1, "maximum": MAX_LIMIT},
},
"required": ["filters"],
},
),
Tool(
name="get_prospect_engagement",
description=(
"Fetch one prospect plus their most recent mailings with delivery, open, click, "
"reply, and bounce timestamps. Use before a call or before deciding whether a "
"prospect has already been contacted."
),
inputSchema={
"type": "object",
"properties": {
"prospect_id": {"type": "string"},
"mailing_limit": {"type": "integer", "minimum": 1, "maximum": 50},
},
"required": ["prospect_id"],
},
),
]
async def _list_sequences(client: httpx.AsyncClient, args: dict[str, Any]) -> Any:
data = await _get(
client, "/sequences", "sequence",
fields=SEQUENCE_FIELDS,
filters=args.get("filters"),
sort="-lastUsedAt",
limit=args.get("limit", DEFAULT_LIMIT),
)
rows = [_flatten(i) for i in data.get("data", []) if _sequence_allowed(i.get("id"))]
return {"count": len(rows), "sequences": rows}
async def _get_sequence_performance(client: httpx.AsyncClient, args: dict[str, Any]) -> Any:
seq_id = str(args["sequence_id"])
if not _sequence_allowed(seq_id):
raise OutreachError(f"sequence {seq_id} is outside OUTREACH_ALLOWED_SEQUENCE_IDS")
data = await _get(client, f"/sequences/{seq_id}", "sequence", fields=SEQUENCE_FIELDS)
item = data.get("data") or {}
attrs = item.get("attributes") or {}
return {"sequence": _flatten(item), "derived": _rates(attrs)}
async def _find_stalled(client: httpx.AsyncClient, args: dict[str, Any]) -> Any:
filters: dict[str, Any] = {"state": args["state"]}
if args.get("sequence_id"):
seq_id = str(args["sequence_id"])
if not _sequence_allowed(seq_id):
raise OutreachError(f"sequence {seq_id} is outside OUTREACH_ALLOWED_SEQUENCE_IDS")
filters["sequence"] = seq_id
data = await _get(
client, "/sequenceStates", "sequenceState",
fields=SEQUENCE_STATE_FIELDS,
filters=filters,
include="prospect,sequence",
sort="-stateChangedAt",
limit=args.get("limit", DEFAULT_LIMIT),
# Included resources carry their own full payload unless projected too. Without
# these two lines every stalled row drags a 230-attribute prospect behind it.
extra_fields={
"prospect": ["firstName", "lastName", "title", "company", "optedOut"],
"sequence": ["name"],
},
)
included = {(i["type"], i["id"]): _flatten(i) for i in data.get("included", [])}
rows = []
for item in data.get("data", []):
row = _flatten(item)
rels = item.get("relationships") or {}
for rel_name in ("prospect", "sequence"):
ref = ((rels.get(rel_name) or {}).get("data")) or {}
if ref:
row[rel_name] = included.get((ref.get("type"), ref.get("id")), {"id": ref.get("id")})
rows.append(row)
return {"state": args["state"], "count": len(rows), "sequence_states": rows}
async def _search_prospects(client: httpx.AsyncClient, args: dict[str, Any]) -> Any:
data = await _get(
client, "/prospects", "prospect",
fields=PROSPECT_FIELDS,
filters=args["filters"],
sort="-touchedAt",
limit=args.get("limit", DEFAULT_LIMIT),
)
rows = [_flatten(i) for i in data.get("data", [])]
contactable = [r for r in rows if not r.get("optedOut") and not r.get("emailOptedOut")]
return {
"count": len(rows),
"contactable_count": len(contactable),
"note": "opt-out flags are not filterable upstream; contactable_count is computed here",
"prospects": rows,
}
async def _get_prospect_engagement(client: httpx.AsyncClient, args: dict[str, Any]) -> Any:
pid = str(args["prospect_id"])
prospect = await _get(client, f"/prospects/{pid}", "prospect", fields=PROSPECT_FIELDS)
mailings = await _get(
client, "/mailings", "mailing",
fields=MAILING_FIELDS,
filters={"prospect": pid},
sort="-createdAt",
limit=args.get("mailing_limit", 10),
)
return {
"prospect": _flatten(prospect.get("data") or {}),
"mailings": [_flatten(i) for i in mailings.get("data", [])],
}
HANDLERS = {
"list_sequences": _list_sequences,
"get_sequence_performance": _get_sequence_performance,
"find_stalled_sequence_states": _find_stalled,
"search_prospects": _search_prospects,
"get_prospect_engagement": _get_prospect_engagement,
}
# ----- Server -----
app = Server("outreach-revops-mcp")
@app.list_tools()
async def list_tools() -> list[Tool]:
return TOOLS
@app.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
handler = HANDLERS.get(name)
if handler is None:
return _ok({"error": f"unknown tool: {name}"})
try:
async with httpx.AsyncClient(timeout=30.0) as client:
return _ok(await handler(client, arguments or {}))
except OutreachError as exc:
# Surfaced as content rather than raised so the model can read the guidance in
# the message (which filter to use, which scope is missing) and correct itself.
return _ok({"error": str(exc)})
except httpx.HTTPError as exc:
return _ok({"error": f"network error talking to Outreach: {exc}"})
async def main() -> None:
TOKENS.load()
async with stdio_server() as (read_stream, write_stream):
await app.run(read_stream, write_stream, app.create_initialization_options())
if __name__ == "__main__":
asyncio.run(main())
