Why this matters
Figma-Context-MCP (published as figma-developer-mcp on npm) gives AI agents direct access to Figma layout metadata rather than relying on screenshots or manual copy-paste [1]. The MCP server pre-processes the Figma API response, stripping irrelevant fields so the model receives only the layout and styling information it needs. That pre-filtering matters: raw Figma API payloads for a moderately complex frame can exceed 50 KB of JSON, most of which is irrelevant to a code-generation task. By the time the MCP server delivers the data, the payload is typically under 5 KB.
LangGraph’s tool-calling loop maps cleanly onto MCP’s request/response model, but wiring the two together requires understanding how LangGraph’s MultiServerMCPClient manages subprocess lifecycles and how to attach OpenTelemetry instrumentation without blocking the event loop. This tutorial builds that wiring from scratch, then adds a synchronous span exporter so you can verify every tool call is traced before connecting to LangSmith or any other OTLP backend.
Prerequisites
- Python 3.11 or 3.12
- Node 18+ with
npxon PATH (the MCP server runs as a Node subprocess) - A Figma personal access token (Settings → Security → Personal access tokens in the Figma web app)
- A Figma file URL containing at least one named frame
- A LangSmith account and API key (used only in the optional tracing section; the core agent runs without it)
- An Anthropic API key (Claude 3 Haiku is used for tool calling; swap for any LangChain-compatible chat model)
Setup
Install the Python dependencies. langgraph brings in langchain-core; langchain-anthropic provides the Claude chat model; langchain-mcp-adapters exposes the MultiServerMCPClient that bridges MCP tools into LangChain tool objects; opentelemetry-sdk provides the tracer and console exporter.
uv pip install langgraph langchain-anthropic "langchain-mcp-adapters>=0.1.0" opentelemetry-sdk opentelemetry-apiVerify the key packages are importable and print their versions:
from importlib.metadata import version
for pkg in ["langgraph", "langchain-anthropic", "langchain-mcp-adapters", "opentelemetry-sdk"]:
print(f"{pkg}: {version(pkg)}")
print("imports OK")Export the credentials you will need. The Figma token and file URL are required to talk to the MCP server. The Anthropic key is required to run the agent. The LangSmith variables are optional and only activate if set.
export FIGMA_API_KEY="your-figma-personal-access-token"
export FIGMA_FILE_URL="https://www.figma.com/file/XXXXXXXXXXXXXXXX/YourFileName"
export ANTHROPIC_API_KEY="your-anthropic-api-key"
# Optional — remove these lines if you don't have a LangSmith account
export LANGCHAIN_TRACING_V2="true"
export LANGCHAIN_API_KEY="your-langsmith-api-key"
export LANGCHAIN_PROJECT="figma-mcp-agent"Step 1: Define the OpenTelemetry tracer
The tracer module sets up a TracerProvider with a SimpleSpanProcessor backed by a ConsoleSpanExporter. Using SimpleSpanProcessor is intentional here: it flushes each span synchronously the moment it ends, so verification blocks can assert on console output without calling force_flush().
In a production pipeline you would swap SimpleSpanProcessor for BatchSpanProcessor and point the exporter at an OTLP endpoint (Grafana Tempo, SigNoz, or any OTLP-compatible backend). The same span structure indexes the same way on Datadog or Honeycomb — only the exporter endpoint changes.
The same span structure indexes the same way on Datadog or Honeycomb — only the exporter endpoint changes.
# filename: otel_setup.py
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor, ConsoleSpanExporter
_provider = TracerProvider()
_provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(_provider)
tracer = trace.get_tracer("figma-mcp-agent")
def get_tracer() -> trace.Tracer:
return tracerStep 2: Build the instrumented tool wrapper
LangGraph tool nodes receive a list of ToolMessage objects. The wrapper below intercepts each tool call, opens an OTel span, records the tool name and a truncated version of the input as span attributes, then closes the span when the call returns. Token counts are not available at the tool level (they come back on the model response), so the wrapper records wall-clock latency instead.
# filename: instrumented_tools.py
import time
import json
from typing import Any
from langchain_core.tools import BaseTool
from otel_setup import get_tracer
def wrap_tool_with_span(tool: BaseTool) -> BaseTool:
"""Return a new tool whose `invoke` method emits an OTel span."""
original_invoke = tool.invoke
tracer = get_tracer()
def traced_invoke(input: Any, config: Any = None, **kwargs: Any) -> Any:
tool_name = tool.name
with tracer.start_as_current_span(f"tool/{tool_name}") as span:
span.set_attribute("tool.name", tool_name)
# Truncate large inputs so the span attribute stays under 512 bytes
raw = json.dumps(input) if not isinstance(input, str) else input
span.set_attribute("tool.input", raw[:512])
start = time.perf_counter()
try:
result = original_invoke(input, config, **kwargs)
span.set_attribute("tool.success", True)
return result
except Exception as exc:
span.set_attribute("tool.success", False)
span.set_attribute("tool.error", str(exc)[:256])
raise
finally:
elapsed_ms = (time.perf_counter() - start) * 1000
span.set_attribute("tool.latency_ms", round(elapsed_ms, 2))
tool.invoke = traced_invoke # type: ignore[method-assign]
return toolStep 3: Define the agent graph
The agent is a standard LangGraph ReAct loop. The model node calls Claude with the bound tools; the tool node executes whichever tool the model selected. The graph routes back to the model after each tool call until the model emits a response with no tool calls.
The build_agent function accepts an optional model argument so the graph structure can be verified in tests without constructing a real API client.
# filename: agent.py
import asyncio
import os
from typing import Annotated, Any
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_core.tools import BaseTool
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from typing_extensions import TypedDict
from instrumented_tools import wrap_tool_with_span
from otel_setup import get_tracer
class AgentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
def build_graph(tools: list[BaseTool], model: Any) -> Any:
"""Compile the ReAct graph given a list of tools and a bound chat model."""
model_with_tools = model.bind_tools(tools)
tool_node = ToolNode(tools)
def call_model(state: AgentState) -> dict:
tracer = get_tracer()
with tracer.start_as_current_span("agent/call_model") as span:
response = model_with_tools.invoke(state["messages"])
span.set_attribute("model.tool_calls", len(getattr(response, "tool_calls", [])))
return {"messages": [response]}
def should_continue(state: AgentState) -> str:
last = state["messages"][-1]
if getattr(last, "tool_calls", None):
return "tools"
return END
graph = StateGraph(AgentState)
graph.add_node("model", call_model)
graph.add_node("tools", tool_node)
graph.set_entry_point("model")
graph.add_conditional_edges("model", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "model")
return graph.compile()
async def run_agent(user_prompt: str) -> str:
"""Connect to the Figma MCP server, run the agent, return the final text."""
figma_key = os.environ["FIGMA_API_KEY"]
mcp_config = {
"figma": {
"command": "npx",
"args": ["-y", "figma-developer-mcp", f"--figma-api-key={figma_key}", "--stdio"],
"transport": "stdio",
}
}
async with MultiServerMCPClient(mcp_config) as client:
raw_tools = client.get_tools()
tools = [wrap_tool_with_span(t) for t in raw_tools]
from langchain_anthropic import ChatAnthropic
model = ChatAnthropic(model="claude-3-haiku-20240307", temperature=0)
app = build_graph(tools, model)
tracer = get_tracer()
with tracer.start_as_current_span("agent/run") as span:
span.set_attribute("agent.prompt", user_prompt[:256])
result = await app.ainvoke({"messages": [HumanMessage(content=user_prompt)]})
final_message = result["messages"][-1]
return final_message.contentStep 4: Verify graph structure without API keys
Before running the live agent, confirm the graph compiles correctly. This block constructs a stub model (no API key needed) and an empty tool list, then prints the node names.
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
class _AgentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
class _StubModel:
def bind_tools(self, tools):
return self
def invoke(self, messages):
from langchain_core.messages import AIMessage
return AIMessage(content="stub")
stub = _StubModel()
tool_node = ToolNode([])
graph = StateGraph(_AgentState)
graph.add_node("model", lambda state: {"messages": [stub.invoke(state["messages"])]})
graph.add_node("tools", tool_node)
graph.set_entry_point("model")
graph.add_edge("tools", "model")
graph.add_conditional_edges("model", lambda s: END, {END: END})
app = graph.compile()
print("Graph nodes:", list(app.get_graph().nodes.keys()))
print("graph_structure_ok")Step 5: Run the live agent against your Figma file
This block is the entry point for the full end-to-end run. It reads FIGMA_FILE_URL from the environment and asks the agent to summarize the top-level frames in the file. The MCP server fetches the Figma API, filters the response, and returns structured layout data [1].
Replace the prompt with any design question relevant to your file, for example: “List all text layers in the Login frame and their font sizes.”
import asyncio
import os
from agent import run_agent
async def main():
file_url = os.environ.get("FIGMA_FILE_URL", "")
if not file_url:
print("FIGMA_FILE_URL not set — skipping live run")
return
prompt = (
f"Using the Figma file at {file_url}, "
"list the names and dimensions of every top-level frame you can find."
)
print("Running agent...")
answer = await run_agent(prompt)
print("\n=== Agent response ===")
print(answer)
asyncio.run(main())Step 6: Inspect the OTel spans
The ConsoleSpanExporter writes each span as JSON to stdout as soon as it closes. A typical tool-call span looks like this (formatted for readability):
{
"name": "tool/get_figma_data",
"context": {"trace_id": "0x...", "span_id": "0x..."},
"attributes": {
"tool.name": "get_figma_data",
"tool.input": "{\"fileKey\": \"XXXXXXXXXXXXXXXX\", \"nodeId\": null}",
"tool.success": true,
"tool.latency_ms": 312.47
},
"start_time": "2024-06-01T12:00:00.000000Z",
"end_time": "2024-06-01T12:00:00.312000Z"
}To forward these spans to LangSmith’s OTLP endpoint instead, replace the ConsoleSpanExporter in otel_setup.py with an OTLPSpanExporter pointed at https://api.smith.langchain.com/otel. LangSmith also captures traces automatically when LANGCHAIN_TRACING_V2=true is set, so for most users the console exporter is sufficient for local debugging and LangSmith handles production observability through its own callback layer.
To forward to a self-hosted Grafana Tempo or SigNoz instance:
# Example — not executed in the sandbox (requires a running OTLP collector)
# uv pip install opentelemetry-exporter-otlp-proto-grpc
# Then in otel_setup.py:
# from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
# exporter = OTLPSpanExporter(endpoint="http://localhost:4317", insecure=True)Verify it works
Run this block to confirm the OTel tracer emits spans synchronously and that the instrumented tool wrapper records latency correctly. No API keys are required.
import time
import io
import sys
from unittest.mock import MagicMock
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor, ConsoleSpanExporter
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
# Use an in-memory exporter so we can assert on spans programmatically
mem_exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(mem_exporter))
trace.set_tracer_provider(provider)
# Build a fake tool
fake_tool = MagicMock()
fake_tool.name = "fake_figma_tool"
fake_tool.invoke = lambda inp, cfg=None, **kw: {"frames": ["Header", "Footer"]}
# Import and apply the wrapper (uses the provider we just set)
from instrumented_tools import wrap_tool_with_span
wrapped = wrap_tool_with_span(fake_tool)
# Invoke the wrapped tool
result = wrapped.invoke({"fileKey": "TEST123"})
assert result == {"frames": ["Header", "Footer"]}, f"Unexpected result: {result}"
# Check the span was recorded
spans = mem_exporter.get_finished_spans()
assert len(spans) == 1, f"Expected 1 span, got {len(spans)}"
span = spans[0]
assert span.name == "tool/fake_figma_tool"
assert span.attributes["tool.name"] == "fake_figma_tool"
assert span.attributes["tool.success"] is True
assert span.attributes["tool.latency_ms"] >= 0
print(f"Span name: {span.name}")
print(f"tool.latency_ms: {span.attributes['tool.latency_ms']}")
print("verify_otel_spans_ok")Troubleshooting
npx: command not found when the MCP client starts the subprocess. The figma-developer-mcp server runs via npx, which ships with Node. Install Node 18+ from nodejs.org or via your system package manager, then confirm npx --version prints a version number.
FIGMA_API_KEY is set but the MCP server returns 403. Figma personal access tokens are scoped. Make sure the token has at least “File content” read access. Regenerate the token in Figma Settings → Security → Personal access tokens and select the correct scopes.
MultiServerMCPClient raises ImportError. The adapter package name on PyPI is langchain-mcp-adapters. Run uv pip install "langchain-mcp-adapters>=0.1.0" and confirm from langchain_mcp_adapters.client import MultiServerMCPClient succeeds.
The agent loops indefinitely without calling any tools. Claude 3 Haiku occasionally fails to call tools when the prompt is ambiguous. Make the prompt more directive: include the exact Figma file URL in the message and explicitly ask the agent to call the Figma tool. If the problem persists, switch to claude-3-5-sonnet-20241022 which has stronger tool-use reliability.
OTel spans appear in stdout but LangSmith shows no traces. LangSmith’s automatic callback tracing (LANGCHAIN_TRACING_V2=true) is separate from the OTLP pipeline. Both can run simultaneously. If LangSmith traces are missing, confirm LANGCHAIN_API_KEY is exported in the same shell session and that the project name in LANGCHAIN_PROJECT matches an existing project in your LangSmith workspace.
asyncio.run() raises RuntimeError: This event loop is already running in a Jupyter notebook. Jupyter runs its own event loop. Replace asyncio.run(main()) with await main() in a notebook cell, or install nest_asyncio and call nest_asyncio.apply() at the top of the notebook.
Next steps
- Emit token usage as span attributes. Claude’s response object carries
usage_metadatawithinput_tokensandoutput_tokens. Add these to theagent/call_modelspan so you can correlate token cost with latency per tool call. - Add a second MCP server.
MultiServerMCPClientaccepts multiple server configs in the same dictionary. Wire in a filesystem MCP server alongside the Figma server so the agent can read the current codebase and write generated components directly to disk. - Persist the graph state with LangGraph’s checkpointer. Add a
SqliteSavercheckpointer tograph.compile()so multi-turn conversations about the same Figma file resume from where they left off without re-fetching layout data. - Gate on frame dimensions. Parse the agent’s response to extract numeric dimensions, then write a pytest fixture that asserts no frame exceeds your design system’s maximum column width. This turns the agent into a lightweight design-lint step in CI.
FAQ
How does the Figma-Context-MCP server reduce payload size?
The MCP server pre-processes Figma API responses, stripping irrelevant fields so the model receives only layout and styling information. Raw Figma API payloads for moderately complex frames often exceed 50 KB, but the MCP server typically delivers data under 5 KB.
What is the purpose of SimpleSpanProcessor in the OpenTelemetry setup?
SimpleSpanProcessor flushes each span synchronously the moment it ends, allowing verification blocks to assert on console output without calling force_flush(). In production, it would be swapped for BatchSpanProcessor pointing to an OTLP endpoint like Grafana Tempo or SigNoz.
How does the instrumented tool wrapper record performance metrics?
The wrapper intercepts each tool call, opens an OTel span, records the tool name and truncated input as span attributes, and measures wall-clock latency using time.perf_counter(). Token counts are not available at the tool level, so latency is recorded instead.
What does MultiServerMCPClient manage in this setup?
MultiServerMCPClient manages subprocess lifecycles for the MCP servers, converts MCP tools into LangChain tool objects, and handles the request/response communication between the LangGraph agent and the Figma MCP server running as a Node subprocess.
Can the same OTel span structure work with different backends?
Yes. The span structure indexes the same way on Datadog, Honeycomb, or any OTLP-compatible backend. Only the exporter endpoint changes; the span attributes and structure remain consistent across observability platforms.