fastmcp.md•240 kB
---
description: Use it to get full context on fastmcp
globs:
alwaysApply: false
---
├── .github
├── ai-labeler.yml
├── release.yml
└── workflows
│ ├── ai-labeler.yml
│ ├── publish.yml
│ ├── run-static.yml
│ └── run-tests.yml
├── .gitignore
├── .pre-commit-config.yaml
├── .python-version
├── LICENSE
├── README.md
├── Windows_Notes.md
├── docs
└── assets
│ └── demo-inspector.png
├── examples
├── complex_inputs.py
├── desktop.py
├── echo.py
├── memory.py
├── readme-quickstart.py
├── screenshot.py
├── simple_echo.py
└── text_me.py
├── pyproject.toml
├── src
└── fastmcp
│ ├── __init__.py
│ ├── cli
│ ├── __init__.py
│ ├── claude.py
│ └── cli.py
│ ├── exceptions.py
│ ├── prompts
│ ├── __init__.py
│ ├── base.py
│ ├── manager.py
│ └── prompt_manager.py
│ ├── py.typed
│ ├── resources
│ ├── __init__.py
│ ├── base.py
│ ├── resource_manager.py
│ ├── templates.py
│ └── types.py
│ ├── server.py
│ ├── tools
│ ├── __init__.py
│ ├── base.py
│ └── tool_manager.py
│ └── utilities
│ ├── __init__.py
│ ├── func_metadata.py
│ ├── logging.py
│ └── types.py
├── tests
├── __init__.py
├── prompts
│ ├── __init__.py
│ ├── test_base.py
│ └── test_manager.py
├── resources
│ ├── __init__.py
│ ├── test_file_resources.py
│ ├── test_function_resources.py
│ ├── test_resource_manager.py
│ ├── test_resource_template.py
│ └── test_resources.py
├── servers
│ ├── __init__.py
│ └── test_file_server.py
├── test_cli.py
├── test_func_metadata.py
├── test_server.py
└── test_tool_manager.py
└── uv.lock
/.github/ai-labeler.yml:
--------------------------------------------------------------------------------
1 | instructions: |
2 | Apply the minimal set of labels that accurately characterize the issue/PR:
3 | - Use at most 1-2 labels unless there's a compelling reason for more. It's ok to use no labels.
4 | - Prefer specific labels (bug, feature) over generic ones (question, help wanted)
5 | - For PRs that fix bugs, use 'bug' not 'enhancement'
6 | - Never combine: bug + enhancement, feature + enhancement. For these labels, only choose the most relevant one.
7 | - Reserve 'question' and 'help wanted' for when they're the primary characteristic
8 |
9 | labels:
10 | - bug:
11 | description: "Something isn't working as expected"
12 | instructions: |
13 | Apply when describing or fixing unexpected behavior:
14 | - Issues: Clear error messages or unexpected outcomes
15 | - PRs: Standalone fixes for broken functionality or closing bug reports.
16 | Don't apply bug unless the issue or PR is predominantly about a specific bug.
17 |
18 | - documentation:
19 | description: "Improvements or additions to documentation"
20 | instructions: |
21 | Apply only when documentation is the primary focus:
22 | - README updates
23 | - Code comments and docstrings
24 | - API documentation
25 | - Usage examples
26 | Don't apply for minor doc updates alongside code changes
27 |
28 | - enhancement:
29 | description: "Improvements to existing features"
30 | instructions: |
31 | Apply only for improvements to existing functionality:
32 | - Performance improvements
33 | - UI/UX improvements
34 | - Expanded capabilities of existing features
35 | Don't apply to:
36 | - Bug fixes
37 | - New features
38 | - Minor tweaks
39 |
40 | - feature:
41 | description: "New functionality"
42 | instructions: |
43 | Apply only for net-new functionality:
44 | - New API endpoints
45 | - New commands or tools
46 | - New user-facing capabilities
47 | Don't apply to:
48 | - Improvements to existing features (use enhancement)
49 | - Bug fixes
50 |
51 | - good first issue:
52 | description: "Good for newcomers"
53 | instructions: |
54 | Apply very selectively to issues that are:
55 | - Small in scope
56 | - Well-documented
57 | - Require minimal context
58 | - Have clear success criteria
59 | Don't apply if the task requires significant background knowledge
60 |
61 | - help wanted:
62 | description: "Extra attention is needed"
63 | instructions: |
64 | Apply only when it's the primary characteristic:
65 | - Issue needs external expertise
66 | - Current maintainers can't address it
67 | - Additional contributors would be valuable
68 | Don't apply just because an issue is open or needs work
69 |
70 | - question:
71 | description: "Further information is requested"
72 | instructions: |
73 | Apply only when the primary purpose is seeking information:
74 | - Clarification needed before work can begin
75 | - Architectural discussions
76 | - Implementation strategy questions
77 | Don't apply to:
78 | - Bug reports that need more details
79 | - Feature requests that need refinement
80 |
81 | # These files will be included in the context if they exist
82 | context-files:
83 | - README.md
84 | - CONTRIBUTING.md
85 | - CODE_OF_CONDUCT.md
86 | - .github/ISSUE_TEMPLATE/bug_report.md
87 | - .github/ISSUE_TEMPLATE/feature_request.md
88 |
--------------------------------------------------------------------------------
/.github/release.yml:
--------------------------------------------------------------------------------
1 | changelog:
2 | exclude:
3 | labels:
4 | - ignore in release notes
5 |
6 | categories:
7 | - title: New Features 🎉
8 | labels:
9 | - feature
10 | - enhancement
11 | exclude:
12 | labels:
13 | - breaking change
14 |
15 | - title: Fixes 🐞
16 | labels:
17 | - bug
18 | exclude:
19 | labels:
20 | - breaking change
21 |
22 | - title: Breaking Changes 🛫
23 | labels:
24 | - breaking change
25 |
26 | - title: Docs 📚
27 | labels:
28 | - documentation
29 |
30 | - title: Other Changes 🦾
31 | labels:
32 | - "*"
33 |
--------------------------------------------------------------------------------
/.github/workflows/ai-labeler.yml:
--------------------------------------------------------------------------------
1 | name: AI Labeler
2 |
3 | on:
4 | issues:
5 | types: [opened, reopened]
6 | issue_comment:
7 | types: [created]
8 | pull_request:
9 | types: [opened, reopened]
10 |
11 | jobs:
12 | ai-labeler:
13 | runs-on: ubuntu-latest
14 | permissions:
15 | contents: read
16 | issues: write
17 | pull-requests: write
18 | steps:
19 | - uses: actions/checkout@v4
20 | - uses: jlowin/ai-labeler@v0.5.0
21 | with:
22 | include-repo-labels: false
23 | openai-api-key: ${{ secrets.OPENAI_API_KEY }}
24 | controlflow-llm-model: openai/gpt-4o-mini
25 |
--------------------------------------------------------------------------------
/.github/workflows/publish.yml:
--------------------------------------------------------------------------------
1 | name: Publish FastMCP to PyPI
2 | on:
3 | release:
4 | types: [published]
5 | workflow_dispatch:
6 |
7 | jobs:
8 | pypi-publish:
9 | name: Upload to PyPI
10 | runs-on: ubuntu-latest
11 | permissions:
12 | id-token: write # For PyPI's trusted publishing
13 | steps:
14 | - name: Checkout
15 | uses: actions/checkout@v4
16 | with:
17 | fetch-depth: 0
18 |
19 | - name: "Install uv"
20 | uses: astral-sh/setup-uv@v3
21 |
22 | - name: Build
23 | run: uv build
24 |
25 | - name: Publish to PyPi
26 | run: uv publish -v dist/*
27 |
--------------------------------------------------------------------------------
/.github/workflows/run-static.yml:
--------------------------------------------------------------------------------
1 | name: Run Pre-commits
2 |
3 | env:
4 | # enable colored output
5 | # https://github.com/pytest-dev/pytest/issues/7443
6 | PY_COLORS: 1
7 |
8 | on:
9 | push:
10 | branches: ["main"]
11 | pull_request:
12 | workflow_dispatch:
13 |
14 | permissions:
15 | contents: read
16 |
17 | jobs:
18 | static_analysis:
19 | timeout-minutes: 1
20 |
21 | runs-on: ubuntu-latest
22 |
23 | steps:
24 | - uses: actions/checkout@v4
25 | - name: Set up Python
26 | uses: actions/setup-python@v5
27 | with:
28 | python-version: "3.12"
29 | - name: Run pre-commit
30 | uses: pre-commit/action@v3.0.1
31 | - name: Install dependencies
32 | run: |
33 | python -m pip install --upgrade pip
34 | pip install ".[tests]"
35 | - name: Run pyright
36 | run: pyright src tests
37 |
--------------------------------------------------------------------------------
/.github/workflows/run-tests.yml:
--------------------------------------------------------------------------------
1 | name: Run tests
2 |
3 | env:
4 | # enable colored output
5 | PY_COLORS: 1
6 |
7 | on:
8 | push:
9 | branches: ["main"]
10 | paths:
11 | - "src/**"
12 | - "tests/**"
13 | - "uv.lock"
14 | - "pyproject.toml"
15 | - ".github/workflows/**"
16 | pull_request:
17 | paths:
18 | - "src/**"
19 | - "tests/**"
20 | - "uv.lock"
21 | - "pyproject.toml"
22 | - ".github/workflows/**"
23 |
24 | workflow_dispatch:
25 |
26 | permissions:
27 | contents: read
28 |
29 | jobs:
30 | run_tests:
31 | name: "Run tests: Python ${{ matrix.python-version }} on ${{ matrix.os }}"
32 | runs-on: ${{ matrix.os }}
33 | strategy:
34 | matrix:
35 | os: [ubuntu-latest, windows-latest, macos-latest]
36 | python-version: ["3.10"]
37 | fail-fast: false
38 |
39 | steps:
40 | - uses: actions/checkout@v4
41 |
42 | - name: Install uv
43 | uses: astral-sh/setup-uv@v4
44 |
45 | - name: Set up Python ${{ matrix.python-version }}
46 | run: uv python install ${{ matrix.python-version }}
47 |
48 | - name: Install FastMCP
49 | run: uv sync --extra tests
50 |
51 | - name: Run tests
52 | run: uv run pytest -vv
53 | if: ${{ !(github.event.pull_request.head.repo.fork) }}
54 |
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
1 | # Python-generated files
2 | __pycache__/
3 | *.py[oc]
4 | build/
5 | dist/
6 | wheels/
7 | *.egg-info
8 |
9 | # Virtual environments
10 | .venv
11 | .DS_Store
12 | .env
13 |
14 |
15 | src/fastmcp/_version.py
16 |
17 | # editors
18 | .cursorrules
19 | .vscode/
20 |
--------------------------------------------------------------------------------
/.pre-commit-config.yaml:
--------------------------------------------------------------------------------
1 | fail_fast: true
2 |
3 | repos:
4 | - repo: https://github.com/abravalheri/validate-pyproject
5 | rev: v0.23
6 | hooks:
7 | - id: validate-pyproject
8 |
9 | - repo: https://github.com/pre-commit/mirrors-prettier
10 | rev: v3.1.0
11 | hooks:
12 | - id: prettier
13 | types_or: [yaml, json5]
14 |
15 | - repo: https://github.com/astral-sh/ruff-pre-commit
16 | rev: v0.8.0
17 | hooks:
18 | - id: ruff-format
19 | - id: ruff
20 | args: [--fix, --exit-non-zero-on-fix]
21 |
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
1 | 3.12
2 |
--------------------------------------------------------------------------------
/LICENSE:
--------------------------------------------------------------------------------
1 | MIT License
2 |
3 | Copyright (c) 2024 Jeremiah Lowin
4 |
5 | Permission is hereby granted, free of charge, to any person obtaining a copy
6 | of this software and associated documentation files (the "Software"), to deal
7 | in the Software without restriction, including without limitation the rights
8 | to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9 | copies of the Software, and to permit persons to whom the Software is
10 | furnished to do so, subject to the following conditions:
11 |
12 | The above copyright notice and this permission notice shall be included in all
13 | copies or substantial portions of the Software.
14 |
15 | THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16 | IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17 | FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18 | AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19 | LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20 | OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
21 | SOFTWARE.
22 |
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
1 | <div align="center">
2 |
3 | ### 🎉 FastMCP has been added to the official MCP SDK! 🎉
4 |
5 | You can now find FastMCP as part of the official Model Context Protocol Python SDK:
6 |
7 | 👉 [github.com/modelcontextprotocol/python-sdk](https://github.com/modelcontextprotocol/python-sdk)
8 |
9 | *Please note: this repository is no longer maintained.*
10 |
11 | ---
12 |
13 |
14 | </br></br></br>
15 |
16 | </div>
17 |
18 | <div align="center">
19 |
20 | <!-- omit in toc -->
21 | # FastMCP 🚀
22 | <strong>The fast, Pythonic way to build MCP servers.</strong>
23 |
24 | [](https://pypi.org/project/fastmcp)
25 | [](https://github.com/jlowin/fastmcp/actions/workflows/run-tests.yml)
26 | [](https://github.com/jlowin/fastmcp/blob/main/LICENSE)
27 |
28 |
29 | </div>
30 |
31 | [Model Context Protocol (MCP)](https://modelcontextprotocol.io) servers are a new, standardized way to provide context and tools to your LLMs, and FastMCP makes building MCP servers simple and intuitive. Create tools, expose resources, and define prompts with clean, Pythonic code:
32 |
33 | ```python
34 | # demo.py
35 |
36 | from fastmcp import FastMCP
37 |
38 |
39 | mcp = FastMCP("Demo 🚀")
40 |
41 |
42 | @mcp.tool()
43 | def add(a: int, b: int) -> int:
44 | """Add two numbers"""
45 | return a + b
46 | ```
47 |
48 | That's it! Give Claude access to the server by running:
49 |
50 | ```bash
51 | fastmcp install demo.py
52 | ```
53 |
54 | FastMCP handles all the complex protocol details and server management, so you can focus on building great tools. It's designed to be high-level and Pythonic - in most cases, decorating a function is all you need.
55 |
56 |
57 | ### Key features:
58 | * **Fast**: High-level interface means less code and faster development
59 | * **Simple**: Build MCP servers with minimal boilerplate
60 | * **Pythonic**: Feels natural to Python developers
61 | * **Complete***: FastMCP aims to provide a full implementation of the core MCP specification
62 |
63 | (\*emphasis on *aims*)
64 |
65 | 🚨 🚧 🏗️ *FastMCP is under active development, as is the MCP specification itself. Core features are working but some advanced capabilities are still in progress.*
66 |
67 |
68 | <!-- omit in toc -->
69 | ## Table of Contents
70 |
71 | - [Installation](#installation)
72 | - [Quickstart](#quickstart)
73 | - [What is MCP?](#what-is-mcp)
74 | - [Core Concepts](#core-concepts)
75 | - [Server](#server)
76 | - [Resources](#resources)
77 | - [Tools](#tools)
78 | - [Prompts](#prompts)
79 | - [Images](#images)
80 | - [Context](#context)
81 | - [Running Your Server](#running-your-server)
82 | - [Development Mode (Recommended for Building \& Testing)](#development-mode-recommended-for-building--testing)
83 | - [Claude Desktop Integration (For Regular Use)](#claude-desktop-integration-for-regular-use)
84 | - [Direct Execution (For Advanced Use Cases)](#direct-execution-for-advanced-use-cases)
85 | - [Server Object Names](#server-object-names)
86 | - [Examples](#examples)
87 | - [Echo Server](#echo-server)
88 | - [SQLite Explorer](#sqlite-explorer)
89 | - [Contributing](#contributing)
90 | - [Prerequisites](#prerequisites)
91 | - [Installation](#installation-1)
92 | - [Testing](#testing)
93 | - [Formatting](#formatting)
94 | - [Opening a Pull Request](#opening-a-pull-request)
95 |
96 | ## Installation
97 |
98 | We strongly recommend installing FastMCP with [uv](https://docs.astral.sh/uv/), as it is required for deploying servers:
99 |
100 | ```bash
101 | uv pip install fastmcp
102 | ```
103 |
104 | Note: on macOS, uv may need to be installed with Homebrew (`brew install uv`) in order to make it available to the Claude Desktop app.
105 |
106 | Alternatively, to use the SDK without deploying, you may use pip:
107 |
108 | ```bash
109 | pip install fastmcp
110 | ```
111 |
112 | ## Quickstart
113 |
114 | Let's create a simple MCP server that exposes a calculator tool and some data:
115 |
116 | ```python
117 | # server.py
118 |
119 | from fastmcp import FastMCP
120 |
121 |
122 | # Create an MCP server
123 | mcp = FastMCP("Demo")
124 |
125 |
126 | # Add an addition tool
127 | @mcp.tool()
128 | def add(a: int, b: int) -> int:
129 | """Add two numbers"""
130 | return a + b
131 |
132 |
133 | # Add a dynamic greeting resource
134 | @mcp.resource("greeting://{name}")
135 | def get_greeting(name: str) -> str:
136 | """Get a personalized greeting"""
137 | return f"Hello, {name}!"
138 | ```
139 |
140 | You can install this server in [Claude Desktop](https://claude.ai/download) and interact with it right away by running:
141 | ```bash
142 | fastmcp install server.py
143 | ```
144 |
145 | Alternatively, you can test it with the MCP Inspector:
146 | ```bash
147 | fastmcp dev server.py
148 | ```
149 |
150 | 
151 |
152 | ## What is MCP?
153 |
154 | The [Model Context Protocol (MCP)](https://modelcontextprotocol.io) lets you build servers that expose data and functionality to LLM applications in a secure, standardized way. Think of it like a web API, but specifically designed for LLM interactions. MCP servers can:
155 |
156 | - Expose data through **Resources** (think of these sort of like GET endpoints; they are used to load information into the LLM's context)
157 | - Provide functionality through **Tools** (sort of like POST endpoints; they are used to execute code or otherwise produce a side effect)
158 | - Define interaction patterns through **Prompts** (reusable templates for LLM interactions)
159 | - And more!
160 |
161 | There is a low-level [Python SDK](https://github.com/modelcontextprotocol/python-sdk) available for implementing the protocol directly, but FastMCP aims to make that easier by providing a high-level, Pythonic interface.
162 |
163 | ## Core Concepts
164 |
165 |
166 | ### Server
167 |
168 | The FastMCP server is your core interface to the MCP protocol. It handles connection management, protocol compliance, and message routing:
169 |
170 | ```python
171 | from fastmcp import FastMCP
172 |
173 | # Create a named server
174 | mcp = FastMCP("My App")
175 |
176 | # Specify dependencies for deployment and development
177 | mcp = FastMCP("My App", dependencies=["pandas", "numpy"])
178 | ```
179 |
180 | ### Resources
181 |
182 | Resources are how you expose data to LLMs. They're similar to GET endpoints in a REST API - they provide data but shouldn't perform significant computation or have side effects. Some examples:
183 |
184 | - File contents
185 | - Database schemas
186 | - API responses
187 | - System information
188 |
189 | Resources can be static:
190 | ```python
191 | @mcp.resource("config://app")
192 | def get_config() -> str:
193 | """Static configuration data"""
194 | return "App configuration here"
195 | ```
196 |
197 | Or dynamic with parameters (FastMCP automatically handles these as MCP templates):
198 | ```python
199 | @mcp.resource("users://{user_id}/profile")
200 | def get_user_profile(user_id: str) -> str:
201 | """Dynamic user data"""
202 | return f"Profile data for user {user_id}"
203 | ```
204 |
205 | ### Tools
206 |
207 | Tools let LLMs take actions through your server. Unlike resources, tools are expected to perform computation and have side effects. They're similar to POST endpoints in a REST API.
208 |
209 | Simple calculation example:
210 | ```python
211 | @mcp.tool()
212 | def calculate_bmi(weight_kg: float, height_m: float) -> float:
213 | """Calculate BMI given weight in kg and height in meters"""
214 | return weight_kg / (height_m ** 2)
215 | ```
216 |
217 | HTTP request example:
218 | ```python
219 | import httpx
220 |
221 | @mcp.tool()
222 | async def fetch_weather(city: str) -> str:
223 | """Fetch current weather for a city"""
224 | async with httpx.AsyncClient() as client:
225 | response = await client.get(
226 | f"https://api.weather.com/{city}"
227 | )
228 | return response.text
229 | ```
230 |
231 | Complex input handling example:
232 | ```python
233 | from pydantic import BaseModel, Field
234 | from typing import Annotated
235 |
236 | class ShrimpTank(BaseModel):
237 | class Shrimp(BaseModel):
238 | name: Annotated[str, Field(max_length=10)]
239 |
240 | shrimp: list[Shrimp]
241 |
242 | @mcp.tool()
243 | def name_shrimp(
244 | tank: ShrimpTank,
245 | # You can use pydantic Field in function signatures for validation.
246 | extra_names: Annotated[list[str], Field(max_length=10)],
247 | ) -> list[str]:
248 | """List all shrimp names in the tank"""
249 | return [shrimp.name for shrimp in tank.shrimp] + extra_names
250 | ```
251 |
252 | ### Prompts
253 |
254 | Prompts are reusable templates that help LLMs interact with your server effectively. They're like "best practices" encoded into your server. A prompt can be as simple as a string:
255 |
256 | ```python
257 | @mcp.prompt()
258 | def review_code(code: str) -> str:
259 | return f"Please review this code:\n\n{code}"
260 | ```
261 |
262 | Or a more structured sequence of messages:
263 | ```python
264 | from fastmcp.prompts.base import UserMessage, AssistantMessage
265 |
266 | @mcp.prompt()
267 | def debug_error(error: str) -> list[Message]:
268 | return [
269 | UserMessage("I'm seeing this error:"),
270 | UserMessage(error),
271 | AssistantMessage("I'll help debug that. What have you tried so far?")
272 | ]
273 | ```
274 |
275 |
276 | ### Images
277 |
278 | FastMCP provides an `Image` class that automatically handles image data in your server:
279 |
280 | ```python
281 | from fastmcp import FastMCP, Image
282 | from PIL import Image as PILImage
283 |
284 | @mcp.tool()
285 | def create_thumbnail(image_path: str) -> Image:
286 | """Create a thumbnail from an image"""
287 | img = PILImage.open(image_path)
288 | img.thumbnail((100, 100))
289 |
290 | # FastMCP automatically handles conversion and MIME types
291 | return Image(data=img.tobytes(), format="png")
292 |
293 | @mcp.tool()
294 | def load_image(path: str) -> Image:
295 | """Load an image from disk"""
296 | # FastMCP handles reading and format detection
297 | return Image(path=path)
298 | ```
299 |
300 | Images can be used as the result of both tools and resources.
301 |
302 | ### Context
303 |
304 | The Context object gives your tools and resources access to MCP capabilities. To use it, add a parameter annotated with `fastmcp.Context`:
305 |
306 | ```python
307 | from fastmcp import FastMCP, Context
308 |
309 | @mcp.tool()
310 | async def long_task(files: list[str], ctx: Context) -> str:
311 | """Process multiple files with progress tracking"""
312 | for i, file in enumerate(files):
313 | ctx.info(f"Processing {file}")
314 | await ctx.report_progress(i, len(files))
315 |
316 | # Read another resource if needed
317 | data = await ctx.read_resource(f"file://{file}")
318 |
319 | return "Processing complete"
320 | ```
321 |
322 | The Context object provides:
323 | - Progress reporting through `report_progress()`
324 | - Logging via `debug()`, `info()`, `warning()`, and `error()`
325 | - Resource access through `read_resource()`
326 | - Request metadata via `request_id` and `client_id`
327 |
328 | ## Running Your Server
329 |
330 | There are three main ways to use your FastMCP server, each suited for different stages of development:
331 |
332 | ### Development Mode (Recommended for Building & Testing)
333 |
334 | The fastest way to test and debug your server is with the MCP Inspector:
335 |
336 | ```bash
337 | fastmcp dev server.py
338 | ```
339 |
340 | This launches a web interface where you can:
341 | - Test your tools and resources interactively
342 | - See detailed logs and error messages
343 | - Monitor server performance
344 | - Set environment variables for testing
345 |
346 | During development, you can:
347 | - Add dependencies with `--with`:
348 | ```bash
349 | fastmcp dev server.py --with pandas --with numpy
350 | ```
351 | - Mount your local code for live updates:
352 | ```bash
353 | fastmcp dev server.py --with-editable .
354 | ```
355 |
356 | ### Claude Desktop Integration (For Regular Use)
357 |
358 | Once your server is ready, install it in Claude Desktop to use it with Claude:
359 |
360 | ```bash
361 | fastmcp install server.py
362 | ```
363 |
364 | Your server will run in an isolated environment with:
365 | - Automatic installation of dependencies specified in your FastMCP instance:
366 | ```python
367 | mcp = FastMCP("My App", dependencies=["pandas", "numpy"])
368 | ```
369 | - Custom naming via `--name`:
370 | ```bash
371 | fastmcp install server.py --name "My Analytics Server"
372 | ```
373 | - Environment variable management:
374 | ```bash
375 | # Set variables individually
376 | fastmcp install server.py -e API_KEY=abc123 -e DB_URL=postgres://...
377 |
378 | # Or load from a .env file
379 | fastmcp install server.py -f .env
380 | ```
381 |
382 | ### Direct Execution (For Advanced Use Cases)
383 |
384 | For advanced scenarios like custom deployments or running without Claude, you can execute your server directly:
385 |
386 | ```python
387 | from fastmcp import FastMCP
388 |
389 | mcp = FastMCP("My App")
390 |
391 | if __name__ == "__main__":
392 | mcp.run()
393 | ```
394 |
395 | Run it with:
396 | ```bash
397 | # Using the FastMCP CLI
398 | fastmcp run server.py
399 |
400 | # Or with Python/uv directly
401 | python server.py
402 | uv run python server.py
403 | ```
404 |
405 |
406 | Note: When running directly, you are responsible for ensuring all dependencies are available in your environment. Any dependencies specified on the FastMCP instance are ignored.
407 |
408 | Choose this method when you need:
409 | - Custom deployment configurations
410 | - Integration with other services
411 | - Direct control over the server lifecycle
412 |
413 | ### Server Object Names
414 |
415 | All FastMCP commands will look for a server object called `mcp`, `app`, or `server` in your file. If you have a different object name or multiple servers in one file, use the syntax `server.py:my_server`:
416 |
417 | ```bash
418 | # Using a standard name
419 | fastmcp run server.py
420 |
421 | # Using a custom name
422 | fastmcp run server.py:my_custom_server
423 | ```
424 |
425 | ## Examples
426 |
427 | Here are a few examples of FastMCP servers. For more, see the `examples/` directory.
428 |
429 | ### Echo Server
430 | A simple server demonstrating resources, tools, and prompts:
431 |
432 | ```python
433 | from fastmcp import FastMCP
434 |
435 | mcp = FastMCP("Echo")
436 |
437 | @mcp.resource("echo://{message}")
438 | def echo_resource(message: str) -> str:
439 | """Echo a message as a resource"""
440 | return f"Resource echo: {message}"
441 |
442 | @mcp.tool()
443 | def echo_tool(message: str) -> str:
444 | """Echo a message as a tool"""
445 | return f"Tool echo: {message}"
446 |
447 | @mcp.prompt()
448 | def echo_prompt(message: str) -> str:
449 | """Create an echo prompt"""
450 | return f"Please process this message: {message}"
451 | ```
452 |
453 | ### SQLite Explorer
454 | A more complex example showing database integration:
455 |
456 | ```python
457 | from fastmcp import FastMCP
458 | import sqlite3
459 |
460 | mcp = FastMCP("SQLite Explorer")
461 |
462 | @mcp.resource("schema://main")
463 | def get_schema() -> str:
464 | """Provide the database schema as a resource"""
465 | conn = sqlite3.connect("database.db")
466 | schema = conn.execute(
467 | "SELECT sql FROM sqlite_master WHERE type='table'"
468 | ).fetchall()
469 | return "\n".join(sql[0] for sql in schema if sql[0])
470 |
471 | @mcp.tool()
472 | def query_data(sql: str) -> str:
473 | """Execute SQL queries safely"""
474 | conn = sqlite3.connect("database.db")
475 | try:
476 | result = conn.execute(sql).fetchall()
477 | return "\n".join(str(row) for row in result)
478 | except Exception as e:
479 | return f"Error: {str(e)}"
480 |
481 | @mcp.prompt()
482 | def analyze_table(table: str) -> str:
483 | """Create a prompt template for analyzing tables"""
484 | return f"""Please analyze this database table:
485 | Table: {table}
486 | Schema:
487 | {get_schema()}
488 |
489 | What insights can you provide about the structure and relationships?"""
490 | ```
491 |
492 | ## Contributing
493 |
494 | <details>
495 |
496 | <summary><h3>Open Developer Guide</h3></summary>
497 |
498 | ### Prerequisites
499 |
500 | FastMCP requires Python 3.10+ and [uv](https://docs.astral.sh/uv/).
501 |
502 | ### Installation
503 |
504 | For development, we recommend installing FastMCP with development dependencies, which includes various utilities the maintainers find useful.
505 |
506 | ```bash
507 | git clone https://github.com/jlowin/fastmcp.git
508 | cd fastmcp
509 | uv sync --frozen --extra dev
510 | ```
511 |
512 | For running tests only (e.g., in CI), you only need the testing dependencies:
513 |
514 | ```bash
515 | uv sync --frozen --extra tests
516 | ```
517 |
518 | ### Testing
519 |
520 | Please make sure to test any new functionality. Your tests should be simple and atomic and anticipate change rather than cement complex patterns.
521 |
522 | Run tests from the root directory:
523 |
524 |
525 | ```bash
526 | pytest -vv
527 | ```
528 |
529 | ### Formatting
530 |
531 | FastMCP enforces a variety of required formats, which you can automatically enforce with pre-commit.
532 |
533 | Install the pre-commit hooks:
534 |
535 | ```bash
536 | pre-commit install
537 | ```
538 |
539 | The hooks will now run on every commit (as well as on every PR). To run them manually:
540 |
541 | ```bash
542 | pre-commit run --all-files
543 | ```
544 |
545 | ### Opening a Pull Request
546 |
547 | Fork the repository and create a new branch:
548 |
549 | ```bash
550 | git checkout -b my-branch
551 | ```
552 |
553 | Make your changes and commit them:
554 |
555 |
556 | ```bash
557 | git add . && git commit -m "My changes"
558 | ```
559 |
560 | Push your changes to your fork:
561 |
562 |
563 | ```bash
564 | git push origin my-branch
565 | ```
566 |
567 | Feel free to reach out in a GitHub issue or discussion if you have any questions!
568 |
569 | </details>
570 |
--------------------------------------------------------------------------------
/Windows_Notes.md:
--------------------------------------------------------------------------------
1 | # Getting your development environment set up properly
2 | To get your environment up and running properly, you'll need a slightly different set of commands that are windows specific:
3 | ```bash
4 | uv venv
5 | .venv\Scripts\activate
6 | uv pip install -e ".[dev]"
7 | ```
8 |
9 | This will install the package in editable mode, and install the development dependencies.
10 |
11 |
12 | # Fixing `AttributeError: module 'collections' has no attribute 'Callable'`
13 | - open `.venv\Lib\site-packages\pyreadline\py3k_compat.py`
14 | - change `return isinstance(x, collections.Callable)` to
15 | ```
16 | from collections.abc import Callable
17 | return isinstance(x, Callable)
18 | ```
19 |
20 | # Helpful notes
21 | For developing FastMCP
22 | ## Install local development version of FastMCP into a local FastMCP project server
23 | - ensure
24 | - change directories to your FastMCP Server location so you can install it in your .venv
25 | - run `.venv\Scripts\activate` to activate your virtual environment
26 | - Then run a series of commands to uninstall the old version and install the new
27 | ```bash
28 | # First uninstall
29 | uv pip uninstall fastmcp
30 |
31 | # Clean any build artifacts in your fastmcp directory
32 | cd C:\path\to\fastmcp
33 | del /s /q *.egg-info
34 |
35 | # Then reinstall in your weather project
36 | cd C:\path\to\new\fastmcp_server
37 | uv pip install --no-cache-dir -e C:\Users\justj\PycharmProjects\fastmcp
38 |
39 | # Check that it installed properly and has the correct git hash
40 | pip show fastmcp
41 | ```
42 |
43 | ## Running the FastMCP server with Inspector
44 | MCP comes with a node.js application called Inspector that can be used to inspect the FastMCP server. To run the inspector, you'll need to install node.js and npm. Then you can run the following commands:
45 | ```bash
46 | fastmcp dev server.py
47 | ```
48 | This will launch a web app on http://localhost:5173/ that you can use to inspect the FastMCP server.
49 |
50 | ## If you start development before creating a fork - your get out of jail free card
51 | - Add your fork as a new remote to your local repository `git remote add fork git@github.com:YOUR-USERNAME/REPOSITORY-NAME.git`
52 | - This will add your repo, short named 'fork', as a remote to your local repository
53 | - Verify that it was added correctly by running `git remote -v`
54 | - Commit your changes
55 | - Push your changes to your fork `git push fork <branch>`
56 | - Create your pull request on GitHub
57 |
58 |
59 |
--------------------------------------------------------------------------------
/docs/assets/demo-inspector.png:
--------------------------------------------------------------------------------
https://raw.githubusercontent.com/jlowin/fastmcp/80c328b3dc0b949f010456ee0e85cc5c90e3305f/docs/assets/demo-inspector.png
--------------------------------------------------------------------------------
/examples/complex_inputs.py:
--------------------------------------------------------------------------------
1 | """
2 | FastMCP Complex inputs Example
3 |
4 | Demonstrates validation via pydantic with complex models.
5 | """
6 |
7 | from pydantic import BaseModel, Field
8 | from typing import Annotated
9 | from fastmcp.server import FastMCP
10 |
11 | mcp = FastMCP("Shrimp Tank")
12 |
13 |
14 | class ShrimpTank(BaseModel):
15 | class Shrimp(BaseModel):
16 | name: Annotated[str, Field(max_length=10)]
17 |
18 | shrimp: list[Shrimp]
19 |
20 |
21 | @mcp.tool()
22 | def name_shrimp(
23 | tank: ShrimpTank,
24 | # You can use pydantic Field in function signatures for validation.
25 | extra_names: Annotated[list[str], Field(max_length=10)],
26 | ) -> list[str]:
27 | """List all shrimp names in the tank"""
28 | return [shrimp.name for shrimp in tank.shrimp] + extra_names
29 |
--------------------------------------------------------------------------------
/examples/desktop.py:
--------------------------------------------------------------------------------
1 | """
2 | FastMCP Desktop Example
3 |
4 | A simple example that exposes the desktop directory as a resource.
5 | """
6 |
7 | from pathlib import Path
8 |
9 | from fastmcp.server import FastMCP
10 |
11 | # Create server
12 | mcp = FastMCP("Demo")
13 |
14 |
15 | @mcp.resource("dir://desktop")
16 | def desktop() -> list[str]:
17 | """List the files in the user's desktop"""
18 | desktop = Path.home() / "Desktop"
19 | return [str(f) for f in desktop.iterdir()]
20 |
21 |
22 | @mcp.tool()
23 | def add(a: int, b: int) -> int:
24 | """Add two numbers"""
25 | return a + b
26 |
--------------------------------------------------------------------------------
/examples/echo.py:
--------------------------------------------------------------------------------
1 | """
2 | FastMCP Echo Server
3 | """
4 |
5 | from fastmcp import FastMCP
6 |
7 | # Create server
8 | mcp = FastMCP("Echo Server")
9 |
10 |
11 | @mcp.tool()
12 | def echo_tool(text: str) -> str:
13 | """Echo the input text"""
14 | return text
15 |
16 |
17 | @mcp.resource("echo://static")
18 | def echo_resource() -> str:
19 | return "Echo!"
20 |
21 |
22 | @mcp.resource("echo://{text}")
23 | def echo_template(text: str) -> str:
24 | """Echo the input text"""
25 | return f"Echo: {text}"
26 |
27 |
28 | @mcp.prompt("echo")
29 | def echo_prompt(text: str) -> str:
30 | return text
31 |
--------------------------------------------------------------------------------
/examples/memory.py:
--------------------------------------------------------------------------------
1 | # /// script
2 | # dependencies = ["pydantic-ai-slim[openai]", "asyncpg", "numpy", "pgvector", "fastmcp"]
3 | # ///
4 |
5 | # uv pip install 'pydantic-ai-slim[openai]' asyncpg numpy pgvector fastmcp
6 |
7 | """
8 | Recursive memory system inspired by the human brain's clustering of memories.
9 | Uses OpenAI's 'text-embedding-3-small' model and pgvector for efficient similarity search.
10 | """
11 |
12 | import asyncio
13 | import math
14 | import os
15 | from dataclasses import dataclass
16 | from datetime import datetime, timezone
17 | from pathlib import Path
18 | from typing import Annotated, Self
19 |
20 | import asyncpg
21 | import numpy as np
22 | from openai import AsyncOpenAI
23 | from pgvector.asyncpg import register_vector # Import register_vector
24 | from pydantic import BaseModel, Field
25 | from pydantic_ai import Agent
26 |
27 | from fastmcp import FastMCP
28 |
29 | MAX_DEPTH = 5
30 | SIMILARITY_THRESHOLD = 0.7
31 | DECAY_FACTOR = 0.99
32 | REINFORCEMENT_FACTOR = 1.1
33 |
34 | DEFAULT_LLM_MODEL = "openai:gpt-4o"
35 | DEFAULT_EMBEDDING_MODEL = "text-embedding-3-small"
36 |
37 | mcp = FastMCP(
38 | "memory",
39 | dependencies=[
40 | "pydantic-ai-slim[openai]",
41 | "asyncpg",
42 | "numpy",
43 | "pgvector",
44 | ],
45 | )
46 |
47 | DB_DSN = "postgresql://postgres:postgres@localhost:54320/memory_db"
48 | # reset memory with rm ~/.fastmcp/{USER}/memory/*
49 | PROFILE_DIR = (
50 | Path.home() / ".fastmcp" / os.environ.get("USER", "anon") / "memory"
51 | ).resolve()
52 | PROFILE_DIR.mkdir(parents=True, exist_ok=True)
53 |
54 |
55 | def cosine_similarity(a: list[float], b: list[float]) -> float:
56 | a_array = np.array(a, dtype=np.float64)
57 | b_array = np.array(b, dtype=np.float64)
58 | return np.dot(a_array, b_array) / (
59 | np.linalg.norm(a_array) * np.linalg.norm(b_array)
60 | )
61 |
62 |
63 | async def do_ai[T](
64 | user_prompt: str,
65 | system_prompt: str,
66 | result_type: type[T] | Annotated,
67 | deps=None,
68 | ) -> T:
69 | agent = Agent(
70 | DEFAULT_LLM_MODEL,
71 | system_prompt=system_prompt,
72 | result_type=result_type,
73 | )
74 | result = await agent.run(user_prompt, deps=deps)
75 | return result.data
76 |
77 |
78 | @dataclass
79 | class Deps:
80 | openai: AsyncOpenAI
81 | pool: asyncpg.Pool
82 |
83 |
84 | async def get_db_pool() -> asyncpg.Pool:
85 | async def init(conn):
86 | await conn.execute("CREATE EXTENSION IF NOT EXISTS vector;")
87 | await register_vector(conn)
88 |
89 | pool = await asyncpg.create_pool(DB_DSN, init=init)
90 | return pool
91 |
92 |
93 | class MemoryNode(BaseModel):
94 | id: int | None = None
95 | content: str
96 | summary: str = ""
97 | importance: float = 1.0
98 | access_count: int = 0
99 | timestamp: float = Field(
100 | default_factory=lambda: datetime.now(timezone.utc).timestamp()
101 | )
102 | embedding: list[float]
103 |
104 | @classmethod
105 | async def from_content(cls, content: str, deps: Deps):
106 | embedding = await get_embedding(content, deps)
107 | return cls(content=content, embedding=embedding)
108 |
109 | async def save(self, deps: Deps):
110 | async with deps.pool.acquire() as conn:
111 | if self.id is None:
112 | result = await conn.fetchrow(
113 | """
114 | INSERT INTO memories (content, summary, importance, access_count, timestamp, embedding)
115 | VALUES ($1, $2, $3, $4, $5, $6)
116 | RETURNING id
117 | """,
118 | self.content,
119 | self.summary,
120 | self.importance,
121 | self.access_count,
122 | self.timestamp,
123 | self.embedding,
124 | )
125 | self.id = result["id"]
126 | else:
127 | await conn.execute(
128 | """
129 | UPDATE memories
130 | SET content = $1, summary = $2, importance = $3,
131 | access_count = $4, timestamp = $5, embedding = $6
132 | WHERE id = $7
133 | """,
134 | self.content,
135 | self.summary,
136 | self.importance,
137 | self.access_count,
138 | self.timestamp,
139 | self.embedding,
140 | self.id,
141 | )
142 |
143 | async def merge_with(self, other: Self, deps: Deps):
144 | self.content = await do_ai(
145 | f"{self.content}\n\n{other.content}",
146 | "Combine the following two texts into a single, coherent text.",
147 | str,
148 | deps,
149 | )
150 | self.importance += other.importance
151 | self.access_count += other.access_count
152 | self.embedding = [(a + b) / 2 for a, b in zip(self.embedding, other.embedding)]
153 | self.summary = await do_ai(
154 | self.content, "Summarize the following text concisely.", str, deps
155 | )
156 | await self.save(deps)
157 | # Delete the merged node from the database
158 | if other.id is not None:
159 | await delete_memory(other.id, deps)
160 |
161 | def get_effective_importance(self):
162 | return self.importance * (1 + math.log(self.access_count + 1))
163 |
164 |
165 | async def get_embedding(text: str, deps: Deps) -> list[float]:
166 | embedding_response = await deps.openai.embeddings.create(
167 | input=text,
168 | model=DEFAULT_EMBEDDING_MODEL,
169 | )
170 | return embedding_response.data[0].embedding
171 |
172 |
173 | async def delete_memory(memory_id: int, deps: Deps):
174 | async with deps.pool.acquire() as conn:
175 | await conn.execute("DELETE FROM memories WHERE id = $1", memory_id)
176 |
177 |
178 | async def add_memory(content: str, deps: Deps):
179 | new_memory = await MemoryNode.from_content(content, deps)
180 | await new_memory.save(deps)
181 |
182 | similar_memories = await find_similar_memories(new_memory.embedding, deps)
183 | for memory in similar_memories:
184 | if memory.id != new_memory.id:
185 | await new_memory.merge_with(memory, deps)
186 |
187 | await update_importance(new_memory.embedding, deps)
188 |
189 | await prune_memories(deps)
190 |
191 | return f"Remembered: {content}"
192 |
193 |
194 | async def find_similar_memories(embedding: list[float], deps: Deps) -> list[MemoryNode]:
195 | async with deps.pool.acquire() as conn:
196 | rows = await conn.fetch(
197 | """
198 | SELECT id, content, summary, importance, access_count, timestamp, embedding
199 | FROM memories
200 | ORDER BY embedding <-> $1
201 | LIMIT 5
202 | """,
203 | embedding,
204 | )
205 | memories = [
206 | MemoryNode(
207 | id=row["id"],
208 | content=row["content"],
209 | summary=row["summary"],
210 | importance=row["importance"],
211 | access_count=row["access_count"],
212 | timestamp=row["timestamp"],
213 | embedding=row["embedding"],
214 | )
215 | for row in rows
216 | ]
217 | return memories
218 |
219 |
220 | async def update_importance(user_embedding: list[float], deps: Deps):
221 | async with deps.pool.acquire() as conn:
222 | rows = await conn.fetch(
223 | "SELECT id, importance, access_count, embedding FROM memories"
224 | )
225 | for row in rows:
226 | memory_embedding = row["embedding"]
227 | similarity = cosine_similarity(user_embedding, memory_embedding)
228 | if similarity > SIMILARITY_THRESHOLD:
229 | new_importance = row["importance"] * REINFORCEMENT_FACTOR
230 | new_access_count = row["access_count"] + 1
231 | else:
232 | new_importance = row["importance"] * DECAY_FACTOR
233 | new_access_count = row["access_count"]
234 | await conn.execute(
235 | """
236 | UPDATE memories
237 | SET importance = $1, access_count = $2
238 | WHERE id = $3
239 | """,
240 | new_importance,
241 | new_access_count,
242 | row["id"],
243 | )
244 |
245 |
246 | async def prune_memories(deps: Deps):
247 | async with deps.pool.acquire() as conn:
248 | rows = await conn.fetch(
249 | """
250 | SELECT id, importance, access_count
251 | FROM memories
252 | ORDER BY importance DESC
253 | OFFSET $1
254 | """,
255 | MAX_DEPTH,
256 | )
257 | for row in rows:
258 | await conn.execute("DELETE FROM memories WHERE id = $1", row["id"])
259 |
260 |
261 | async def display_memory_tree(deps: Deps) -> str:
262 | async with deps.pool.acquire() as conn:
263 | rows = await conn.fetch(
264 | """
265 | SELECT content, summary, importance, access_count
266 | FROM memories
267 | ORDER BY importance DESC
268 | LIMIT $1
269 | """,
270 | MAX_DEPTH,
271 | )
272 | result = ""
273 | for row in rows:
274 | effective_importance = row["importance"] * (
275 | 1 + math.log(row["access_count"] + 1)
276 | )
277 | summary = row["summary"] or row["content"]
278 | result += f"- {summary} (Importance: {effective_importance:.2f})\n"
279 | return result
280 |
281 |
282 | @mcp.tool()
283 | async def remember(
284 | contents: list[str] = Field(
285 | description="List of observations or memories to store"
286 | ),
287 | ):
288 | deps = Deps(openai=AsyncOpenAI(), pool=await get_db_pool())
289 | try:
290 | return "\n".join(
291 | await asyncio.gather(*[add_memory(content, deps) for content in contents])
292 | )
293 | finally:
294 | await deps.pool.close()
295 |
296 |
297 | @mcp.tool()
298 | async def read_profile() -> str:
299 | deps = Deps(openai=AsyncOpenAI(), pool=await get_db_pool())
300 | profile = await display_memory_tree(deps)
301 | await deps.pool.close()
302 | return profile
303 |
304 |
305 | async def initialize_database():
306 | pool = await asyncpg.create_pool(
307 | "postgresql://postgres:postgres@localhost:54320/postgres"
308 | )
309 | try:
310 | async with pool.acquire() as conn:
311 | await conn.execute("""
312 | SELECT pg_terminate_backend(pg_stat_activity.pid)
313 | FROM pg_stat_activity
314 | WHERE pg_stat_activity.datname = 'memory_db'
315 | AND pid <> pg_backend_pid();
316 | """)
317 | await conn.execute("DROP DATABASE IF EXISTS memory_db;")
318 | await conn.execute("CREATE DATABASE memory_db;")
319 | finally:
320 | await pool.close()
321 |
322 | pool = await asyncpg.create_pool(DB_DSN)
323 | try:
324 | async with pool.acquire() as conn:
325 | await conn.execute("CREATE EXTENSION IF NOT EXISTS vector;")
326 |
327 | await register_vector(conn)
328 |
329 | await conn.execute("""
330 | CREATE TABLE IF NOT EXISTS memories (
331 | id SERIAL PRIMARY KEY,
332 | content TEXT NOT NULL,
333 | summary TEXT,
334 | importance REAL NOT NULL,
335 | access_count INT NOT NULL,
336 | timestamp DOUBLE PRECISION NOT NULL,
337 | embedding vector(1536) NOT NULL
338 | );
339 | CREATE INDEX IF NOT EXISTS idx_memories_embedding ON memories USING hnsw (embedding vector_l2_ops);
340 | """)
341 | finally:
342 | await pool.close()
343 |
344 |
345 | if __name__ == "__main__":
346 | asyncio.run(initialize_database())
347 |
--------------------------------------------------------------------------------
/examples/readme-quickstart.py:
--------------------------------------------------------------------------------
1 | from fastmcp import FastMCP
2 |
3 |
4 | # Create an MCP server
5 | mcp = FastMCP("Demo")
6 |
7 |
8 | # Add an addition tool
9 | @mcp.tool()
10 | def add(a: int, b: int) -> int:
11 | """Add two numbers"""
12 | return a + b
13 |
14 |
15 | # Add a dynamic greeting resource
16 | @mcp.resource("greeting://{name}")
17 | def get_greeting(name: str) -> str:
18 | """Get a personalized greeting"""
19 | return f"Hello, {name}!"
20 |
--------------------------------------------------------------------------------
/examples/screenshot.py:
--------------------------------------------------------------------------------
1 | """
2 | FastMCP Screenshot Example
3 |
4 | Give Claude a tool to capture and view screenshots.
5 | """
6 |
7 | import io
8 | from fastmcp import FastMCP, Image
9 |
10 |
11 | # Create server
12 | mcp = FastMCP("Screenshot Demo", dependencies=["pyautogui", "Pillow"])
13 |
14 |
15 | @mcp.tool()
16 | def take_screenshot() -> Image:
17 | """
18 | Take a screenshot of the user's screen and return it as an image. Use
19 | this tool anytime the user wants you to look at something they're doing.
20 | """
21 | import pyautogui
22 |
23 | buffer = io.BytesIO()
24 |
25 | # if the file exceeds ~1MB, it will be rejected by Claude
26 | screenshot = pyautogui.screenshot()
27 | screenshot.convert("RGB").save(buffer, format="JPEG", quality=60, optimize=True)
28 | return Image(data=buffer.getvalue(), format="jpeg")
29 |
--------------------------------------------------------------------------------
/examples/simple_echo.py:
--------------------------------------------------------------------------------
1 | """
2 | FastMCP Echo Server
3 | """
4 |
5 | from fastmcp import FastMCP
6 |
7 |
8 | # Create server
9 | mcp = FastMCP("Echo Server")
10 |
11 |
12 | @mcp.tool()
13 | def echo(text: str) -> str:
14 | """Echo the input text"""
15 | return text
16 |
--------------------------------------------------------------------------------
/examples/text_me.py:
--------------------------------------------------------------------------------
1 | # /// script
2 | # dependencies = ["fastmcp"]
3 | # ///
4 |
5 | """
6 | FastMCP Text Me Server
7 | --------------------------------
8 | This defines a simple FastMCP server that sends a text message to a phone number via https://surgemsg.com/.
9 |
10 | To run this example, create a `.env` file with the following values:
11 |
12 | SURGE_API_KEY=...
13 | SURGE_ACCOUNT_ID=...
14 | SURGE_MY_PHONE_NUMBER=...
15 | SURGE_MY_FIRST_NAME=...
16 | SURGE_MY_LAST_NAME=...
17 |
18 | Visit https://surgemsg.com/ and click "Get Started" to obtain these values.
19 | """
20 |
21 | from typing import Annotated
22 | import httpx
23 | from pydantic import BeforeValidator
24 | from pydantic_settings import BaseSettings, SettingsConfigDict
25 |
26 | from fastmcp import FastMCP
27 |
28 |
29 | class SurgeSettings(BaseSettings):
30 | model_config: SettingsConfigDict = SettingsConfigDict(
31 | env_prefix="SURGE_", env_file=".env"
32 | )
33 |
34 | api_key: str
35 | account_id: str
36 | my_phone_number: Annotated[
37 | str, BeforeValidator(lambda v: "+" + v if not v.startswith("+") else v)
38 | ]
39 | my_first_name: str
40 | my_last_name: str
41 |
42 |
43 | # Create server
44 | mcp = FastMCP("Text me")
45 | surge_settings = SurgeSettings() # type: ignore
46 |
47 |
48 | @mcp.tool(name="textme", description="Send a text message to me")
49 | def text_me(text_content: str) -> str:
50 | """Send a text message to a phone number via https://surgemsg.com/"""
51 | with httpx.Client() as client:
52 | response = client.post(
53 | "https://api.surgemsg.com/messages",
54 | headers={
55 | "Authorization": f"Bearer {surge_settings.api_key}",
56 | "Surge-Account": surge_settings.account_id,
57 | "Content-Type": "application/json",
58 | },
59 | json={
60 | "body": text_content,
61 | "conversation": {
62 | "contact": {
63 | "first_name": surge_settings.my_first_name,
64 | "last_name": surge_settings.my_last_name,
65 | "phone_number": surge_settings.my_phone_number,
66 | }
67 | },
68 | },
69 | )
70 | response.raise_for_status()
71 | return f"Message sent: {text_content}"
72 |
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
1 | [project]
2 | name = "fastmcp"
3 | dynamic = ["version"]
4 | description = "A more ergonomic interface for MCP servers"
5 | authors = [{ name = "Jeremiah Lowin" }]
6 | dependencies = [
7 | "httpx>=0.26.0",
8 | "mcp>=1.0.0,<2.0.0",
9 | "pydantic-settings>=2.6.1",
10 | "pydantic>=2.5.3,<3.0.0",
11 | "typer>=0.9.0",
12 | "python-dotenv>=1.0.1",
13 | ]
14 | requires-python = ">=3.10"
15 | readme = "README.md"
16 | license = { text = "MIT" }
17 |
18 | [project.scripts]
19 | fastmcp = "fastmcp.cli:app"
20 |
21 | [build-system]
22 | requires = ["hatchling>=1.21.0", "hatch-vcs>=0.4.0"]
23 | build-backend = "hatchling.build"
24 |
25 | [project.optional-dependencies]
26 | tests = [
27 | "pre-commit",
28 | "pyright>=1.1.389",
29 | "pytest>=8.3.3",
30 | "pytest-asyncio>=0.23.5",
31 | "pytest-flakefinder",
32 | "pytest-xdist>=3.6.1",
33 | "ruff",
34 | ]
35 | dev = ["fastmcp[tests]", "copychat>=0.5.2", "ipython>=8.12.3", "pdbpp>=0.10.3"]
36 |
37 | [tool.pytest.ini_options]
38 | asyncio_mode = "auto"
39 | asyncio_default_fixture_loop_scope = "session"
40 |
41 | [tool.hatch.version]
42 | source = "vcs"
43 |
44 | [tool.pyright]
45 | include = ["src", "tests"]
46 | exclude = ["**/node_modules", "**/__pycache__", ".venv", ".git", "dist"]
47 | pythonVersion = "3.10"
48 | pythonPlatform = "Darwin"
49 | typeCheckingMode = "basic"
50 | reportMissingImports = true
51 | reportMissingTypeStubs = false
52 | useLibraryCodeForTypes = true
53 | venvPath = "."
54 | venv = ".venv"
55 |
--------------------------------------------------------------------------------
/src/fastmcp/__init__.py:
--------------------------------------------------------------------------------
1 | """FastMCP - A more ergonomic interface for MCP servers."""
2 |
3 | from importlib.metadata import version
4 | from .server import FastMCP, Context
5 | from .utilities.types import Image
6 |
7 | __version__ = version("fastmcp")
8 | __all__ = ["FastMCP", "Context", "Image"]
9 |
--------------------------------------------------------------------------------
/src/fastmcp/cli/__init__.py:
--------------------------------------------------------------------------------
1 | """FastMCP CLI package."""
2 |
3 | from .cli import app
4 |
5 |
6 | if __name__ == "__main__":
7 | app()
8 |
--------------------------------------------------------------------------------
/src/fastmcp/cli/claude.py:
--------------------------------------------------------------------------------
1 | """Claude app integration utilities."""
2 |
3 | import json
4 | import sys
5 | from pathlib import Path
6 | from typing import Optional, Dict
7 |
8 | from ..utilities.logging import get_logger
9 |
10 | logger = get_logger(__name__)
11 |
12 |
13 | def get_claude_config_path() -> Path | None:
14 | """Get the Claude config directory based on platform."""
15 | if sys.platform == "win32":
16 | path = Path(Path.home(), "AppData", "Roaming", "Claude")
17 | elif sys.platform == "darwin":
18 | path = Path(Path.home(), "Library", "Application Support", "Claude")
19 | else:
20 | return None
21 |
22 | if path.exists():
23 | return path
24 | return None
25 |
26 |
27 | def update_claude_config(
28 | file_spec: str,
29 | server_name: str,
30 | *,
31 | with_editable: Optional[Path] = None,
32 | with_packages: Optional[list[str]] = None,
33 | env_vars: Optional[Dict[str, str]] = None,
34 | ) -> bool:
35 | """Add or update a FastMCP server in Claude's configuration.
36 |
37 | Args:
38 | file_spec: Path to the server file, optionally with :object suffix
39 | server_name: Name for the server in Claude's config
40 | with_editable: Optional directory to install in editable mode
41 | with_packages: Optional list of additional packages to install
42 | env_vars: Optional dictionary of environment variables. These are merged with
43 | any existing variables, with new values taking precedence.
44 |
45 | Raises:
46 | RuntimeError: If Claude Desktop's config directory is not found, indicating
47 | Claude Desktop may not be installed or properly set up.
48 | """
49 | config_dir = get_claude_config_path()
50 | if not config_dir:
51 | raise RuntimeError(
52 | "Claude Desktop config directory not found. Please ensure Claude Desktop "
53 | "is installed and has been run at least once to initialize its configuration."
54 | )
55 |
56 | config_file = config_dir / "claude_desktop_config.json"
57 | if not config_file.exists():
58 | try:
59 | config_file.write_text("{}")
60 | except Exception as e:
61 | logger.error(
62 | "Failed to create Claude config file",
63 | extra={
64 | "error": str(e),
65 | "config_file": str(config_file),
66 | },
67 | )
68 | return False
69 |
70 | try:
71 | config = json.loads(config_file.read_text())
72 | if "mcpServers" not in config:
73 | config["mcpServers"] = {}
74 |
75 | # Always preserve existing env vars and merge with new ones
76 | if (
77 | server_name in config["mcpServers"]
78 | and "env" in config["mcpServers"][server_name]
79 | ):
80 | existing_env = config["mcpServers"][server_name]["env"]
81 | if env_vars:
82 | # New vars take precedence over existing ones
83 | env_vars = {**existing_env, **env_vars}
84 | else:
85 | env_vars = existing_env
86 |
87 | # Build uv run command
88 | args = ["run"]
89 |
90 | # Collect all packages in a set to deduplicate
91 | packages = {"fastmcp"}
92 | if with_packages:
93 | packages.update(pkg for pkg in with_packages if pkg)
94 |
95 | # Add all packages with --with
96 | for pkg in sorted(packages):
97 | args.extend(["--with", pkg])
98 |
99 | if with_editable:
100 | args.extend(["--with-editable", str(with_editable)])
101 |
102 | # Convert file path to absolute before adding to command
103 | # Split off any :object suffix first
104 | if ":" in file_spec:
105 | file_path, server_object = file_spec.rsplit(":", 1)
106 | file_spec = f"{Path(file_path).resolve()}:{server_object}"
107 | else:
108 | file_spec = str(Path(file_spec).resolve())
109 |
110 | # Add fastmcp run command
111 | args.extend(["fastmcp", "run", file_spec])
112 |
113 | server_config = {
114 | "command": "uv",
115 | "args": args,
116 | }
117 |
118 | # Add environment variables if specified
119 | if env_vars:
120 | server_config["env"] = env_vars
121 |
122 | config["mcpServers"][server_name] = server_config
123 |
124 | config_file.write_text(json.dumps(config, indent=2))
125 | logger.info(
126 | f"Added server '{server_name}' to Claude config",
127 | extra={"config_file": str(config_file)},
128 | )
129 | return True
130 | except Exception as e:
131 | logger.error(
132 | "Failed to update Claude config",
133 | extra={
134 | "error": str(e),
135 | "config_file": str(config_file),
136 | },
137 | )
138 | return False
139 |
--------------------------------------------------------------------------------
/src/fastmcp/cli/cli.py:
--------------------------------------------------------------------------------
1 | """FastMCP CLI tools."""
2 |
3 | import importlib.metadata
4 | import importlib.util
5 | import os
6 | import subprocess
7 | import sys
8 | from pathlib import Path
9 | from typing import Dict, Optional, Tuple
10 |
11 | import dotenv
12 | import typer
13 | from typing_extensions import Annotated
14 |
15 | from fastmcp.cli import claude
16 | from fastmcp.utilities.logging import get_logger
17 |
18 | logger = get_logger("cli")
19 |
20 | app = typer.Typer(
21 | name="fastmcp",
22 | help="FastMCP development tools",
23 | add_completion=False,
24 | no_args_is_help=True, # Show help if no args provided
25 | )
26 |
27 |
28 | def _get_npx_command():
29 | """Get the correct npx command for the current platform."""
30 | if sys.platform == "win32":
31 | # Try both npx.cmd and npx.exe on Windows
32 | for cmd in ["npx.cmd", "npx.exe", "npx"]:
33 | try:
34 | subprocess.run(
35 | [cmd, "--version"], check=True, capture_output=True, shell=True
36 | )
37 | return cmd
38 | except subprocess.CalledProcessError:
39 | continue
40 | return None
41 | return "npx" # On Unix-like systems, just use npx
42 |
43 |
44 | def _parse_env_var(env_var: str) -> Tuple[str, str]:
45 | """Parse environment variable string in format KEY=VALUE."""
46 | if "=" not in env_var:
47 | logger.error(
48 | f"Invalid environment variable format: {env_var}. Must be KEY=VALUE"
49 | )
50 | sys.exit(1)
51 | key, value = env_var.split("=", 1)
52 | return key.strip(), value.strip()
53 |
54 |
55 | def _build_uv_command(
56 | file_spec: str,
57 | with_editable: Optional[Path] = None,
58 | with_packages: Optional[list[str]] = None,
59 | ) -> list[str]:
60 | """Build the uv run command that runs a FastMCP server through fastmcp run."""
61 | cmd = ["uv"]
62 |
63 | cmd.extend(["run", "--with", "fastmcp"])
64 |
65 | if with_editable:
66 | cmd.extend(["--with-editable", str(with_editable)])
67 |
68 | if with_packages:
69 | for pkg in with_packages:
70 | if pkg:
71 | cmd.extend(["--with", pkg])
72 |
73 | # Add fastmcp run command
74 | cmd.extend(["fastmcp", "run", file_spec])
75 | return cmd
76 |
77 |
78 | def _parse_file_path(file_spec: str) -> Tuple[Path, Optional[str]]:
79 | """Parse a file path that may include a server object specification.
80 |
81 | Args:
82 | file_spec: Path to file, optionally with :object suffix
83 |
84 | Returns:
85 | Tuple of (file_path, server_object)
86 | """
87 | # First check if we have a Windows path (e.g., C:\...)
88 | has_windows_drive = len(file_spec) > 1 and file_spec[1] == ":"
89 |
90 | # Split on the last colon, but only if it's not part of the Windows drive letter
91 | # and there's actually another colon in the string after the drive letter
92 | if ":" in (file_spec[2:] if has_windows_drive else file_spec):
93 | file_str, server_object = file_spec.rsplit(":", 1)
94 | else:
95 | file_str, server_object = file_spec, None
96 |
97 | # Resolve the file path
98 | file_path = Path(file_str).expanduser().resolve()
99 | if not file_path.exists():
100 | logger.error(f"File not found: {file_path}")
101 | sys.exit(1)
102 | if not file_path.is_file():
103 | logger.error(f"Not a file: {file_path}")
104 | sys.exit(1)
105 |
106 | return file_path, server_object
107 |
108 |
109 | def _import_server(file: Path, server_object: Optional[str] = None):
110 | """Import a FastMCP server from a file.
111 |
112 | Args:
113 | file: Path to the file
114 | server_object: Optional object name in format "module:object" or just "object"
115 |
116 | Returns:
117 | The server object
118 | """
119 | # Add parent directory to Python path so imports can be resolved
120 | file_dir = str(file.parent)
121 | if file_dir not in sys.path:
122 | sys.path.insert(0, file_dir)
123 |
124 | # Import the module
125 | spec = importlib.util.spec_from_file_location("server_module", file)
126 | if not spec or not spec.loader:
127 | logger.error("Could not load module", extra={"file": str(file)})
128 | sys.exit(1)
129 |
130 | module = importlib.util.module_from_spec(spec)
131 | spec.loader.exec_module(module)
132 |
133 | # If no object specified, try common server names
134 | if not server_object:
135 | # Look for the most common server object names
136 | for name in ["mcp", "server", "app"]:
137 | if hasattr(module, name):
138 | return getattr(module, name)
139 |
140 | logger.error(
141 | f"No server object found in {file}. Please either:\n"
142 | "1. Use a standard variable name (mcp, server, or app)\n"
143 | "2. Specify the object name with file:object syntax",
144 | extra={"file": str(file)},
145 | )
146 | sys.exit(1)
147 |
148 | # Handle module:object syntax
149 | if ":" in server_object:
150 | module_name, object_name = server_object.split(":", 1)
151 | try:
152 | server_module = importlib.import_module(module_name)
153 | server = getattr(server_module, object_name, None)
154 | except ImportError:
155 | logger.error(
156 | f"Could not import module '{module_name}'",
157 | extra={"file": str(file)},
158 | )
159 | sys.exit(1)
160 | else:
161 | # Just object name
162 | server = getattr(module, server_object, None)
163 |
164 | if server is None:
165 | logger.error(
166 | f"Server object '{server_object}' not found",
167 | extra={"file": str(file)},
168 | )
169 | sys.exit(1)
170 |
171 | return server
172 |
173 |
174 | @app.command()
175 | def version() -> None:
176 | """Show the FastMCP version."""
177 | try:
178 | version = importlib.metadata.version("fastmcp")
179 | print(f"FastMCP version {version}")
180 | except importlib.metadata.PackageNotFoundError:
181 | print("FastMCP version unknown (package not installed)")
182 | sys.exit(1)
183 |
184 |
185 | @app.command()
186 | def dev(
187 | file_spec: str = typer.Argument(
188 | ...,
189 | help="Python file to run, optionally with :object suffix",
190 | ),
191 | with_editable: Annotated[
192 | Optional[Path],
193 | typer.Option(
194 | "--with-editable",
195 | "-e",
196 | help="Directory containing pyproject.toml to install in editable mode",
197 | exists=True,
198 | file_okay=False,
199 | resolve_path=True,
200 | ),
201 | ] = None,
202 | with_packages: Annotated[
203 | list[str],
204 | typer.Option(
205 | "--with",
206 | help="Additional packages to install",
207 | ),
208 | ] = [],
209 | ) -> None:
210 | """Run a FastMCP server with the MCP Inspector."""
211 | file, server_object = _parse_file_path(file_spec)
212 |
213 | logger.debug(
214 | "Starting dev server",
215 | extra={
216 | "file": str(file),
217 | "server_object": server_object,
218 | "with_editable": str(with_editable) if with_editable else None,
219 | "with_packages": with_packages,
220 | },
221 | )
222 |
223 | try:
224 | # Import server to get dependencies
225 | server = _import_server(file, server_object)
226 | if hasattr(server, "dependencies"):
227 | with_packages = list(set(with_packages + server.dependencies))
228 |
229 | uv_cmd = _build_uv_command(file_spec, with_editable, with_packages)
230 |
231 | # Get the correct npx command
232 | npx_cmd = _get_npx_command()
233 | if not npx_cmd:
234 | logger.error(
235 | "npx not found. Please ensure Node.js and npm are properly installed "
236 | "and added to your system PATH."
237 | )
238 | sys.exit(1)
239 |
240 | # Run the MCP Inspector command with shell=True on Windows
241 | shell = sys.platform == "win32"
242 | process = subprocess.run(
243 | [npx_cmd, "@modelcontextprotocol/inspector"] + uv_cmd,
244 | check=True,
245 | shell=shell,
246 | env=dict(os.environ.items()), # Convert to list of tuples for env update
247 | )
248 | sys.exit(process.returncode)
249 | except subprocess.CalledProcessError as e:
250 | logger.error(
251 | "Dev server failed",
252 | extra={
253 | "file": str(file),
254 | "error": str(e),
255 | "returncode": e.returncode,
256 | },
257 | )
258 | sys.exit(e.returncode)
259 | except FileNotFoundError:
260 | logger.error(
261 | "npx not found. Please ensure Node.js and npm are properly installed "
262 | "and added to your system PATH. You may need to restart your terminal "
263 | "after installation.",
264 | extra={"file": str(file)},
265 | )
266 | sys.exit(1)
267 |
268 |
269 | @app.command()
270 | def run(
271 | file_spec: str = typer.Argument(
272 | ...,
273 | help="Python file to run, optionally with :object suffix",
274 | ),
275 | transport: Annotated[
276 | Optional[str],
277 | typer.Option(
278 | "--transport",
279 | "-t",
280 | help="Transport protocol to use (stdio or sse)",
281 | ),
282 | ] = None,
283 | ) -> None:
284 | """Run a FastMCP server.
285 |
286 | The server can be specified in two ways:
287 | 1. Module approach: server.py - runs the module directly, expecting a server.run() call
288 | 2. Import approach: server.py:app - imports and runs the specified server object
289 |
290 | Note: This command runs the server directly. You are responsible for ensuring
291 | all dependencies are available. For dependency management, use fastmcp install
292 | or fastmcp dev instead.
293 | """
294 | file, server_object = _parse_file_path(file_spec)
295 |
296 | logger.debug(
297 | "Running server",
298 | extra={
299 | "file": str(file),
300 | "server_object": server_object,
301 | "transport": transport,
302 | },
303 | )
304 |
305 | try:
306 | # Import and get server object
307 | server = _import_server(file, server_object)
308 |
309 | # Run the server
310 | kwargs = {}
311 | if transport:
312 | kwargs["transport"] = transport
313 |
314 | server.run(**kwargs)
315 |
316 | except Exception as e:
317 | logger.error(
318 | f"Failed to run server: {e}",
319 | extra={
320 | "file": str(file),
321 | "error": str(e),
322 | },
323 | )
324 | sys.exit(1)
325 |
326 |
327 | @app.command()
328 | def install(
329 | file_spec: str = typer.Argument(
330 | ...,
331 | help="Python file to run, optionally with :object suffix",
332 | ),
333 | server_name: Annotated[
334 | Optional[str],
335 | typer.Option(
336 | "--name",
337 | "-n",
338 | help="Custom name for the server (defaults to server's name attribute or file name)",
339 | ),
340 | ] = None,
341 | with_editable: Annotated[
342 | Optional[Path],
343 | typer.Option(
344 | "--with-editable",
345 | "-e",
346 | help="Directory containing pyproject.toml to install in editable mode",
347 | exists=True,
348 | file_okay=False,
349 | resolve_path=True,
350 | ),
351 | ] = None,
352 | with_packages: Annotated[
353 | list[str],
354 | typer.Option(
355 | "--with",
356 | help="Additional packages to install",
357 | ),
358 | ] = [],
359 | env_vars: Annotated[
360 | list[str],
361 | typer.Option(
362 | "--env-var",
363 | "-e",
364 | help="Environment variables in KEY=VALUE format",
365 | ),
366 | ] = [],
367 | env_file: Annotated[
368 | Optional[Path],
369 | typer.Option(
370 | "--env-file",
371 | "-f",
372 | help="Load environment variables from a .env file",
373 | exists=True,
374 | file_okay=True,
375 | dir_okay=False,
376 | resolve_path=True,
377 | ),
378 | ] = None,
379 | ) -> None:
380 | """Install a FastMCP server in the Claude desktop app.
381 |
382 | Environment variables are preserved once added and only updated if new values
383 | are explicitly provided.
384 | """
385 | file, server_object = _parse_file_path(file_spec)
386 |
387 | logger.debug(
388 | "Installing server",
389 | extra={
390 | "file": str(file),
391 | "server_name": server_name,
392 | "server_object": server_object,
393 | "with_editable": str(with_editable) if with_editable else None,
394 | "with_packages": with_packages,
395 | },
396 | )
397 |
398 | if not claude.get_claude_config_path():
399 | logger.error("Claude app not found")
400 | sys.exit(1)
401 |
402 | # Try to import server to get its name, but fall back to file name if dependencies missing
403 | name = server_name
404 | server = None
405 | if not name:
406 | try:
407 | server = _import_server(file, server_object)
408 | name = server.name
409 | except (ImportError, ModuleNotFoundError) as e:
410 | logger.debug(
411 | "Could not import server (likely missing dependencies), using file name",
412 | extra={"error": str(e)},
413 | )
414 | name = file.stem
415 |
416 | # Get server dependencies if available
417 | server_dependencies = getattr(server, "dependencies", []) if server else []
418 | if server_dependencies:
419 | with_packages = list(set(with_packages + server_dependencies))
420 |
421 | # Process environment variables if provided
422 | env_dict: Optional[Dict[str, str]] = None
423 | if env_file or env_vars:
424 | env_dict = {}
425 | # Load from .env file if specified
426 | if env_file:
427 | try:
428 | env_dict |= {
429 | k: v
430 | for k, v in dotenv.dotenv_values(env_file).items()
431 | if v is not None
432 | }
433 | except Exception as e:
434 | logger.error(f"Failed to load .env file: {e}")
435 | sys.exit(1)
436 |
437 | # Add command line environment variables
438 | for env_var in env_vars:
439 | key, value = _parse_env_var(env_var)
440 | env_dict[key] = value
441 |
442 | if claude.update_claude_config(
443 | file_spec,
444 | name,
445 | with_editable=with_editable,
446 | with_packages=with_packages,
447 | env_vars=env_dict,
448 | ):
449 | logger.info(f"Successfully installed {name} in Claude app")
450 | else:
451 | logger.error(f"Failed to install {name} in Claude app")
452 | sys.exit(1)
453 |
--------------------------------------------------------------------------------
/src/fastmcp/exceptions.py:
--------------------------------------------------------------------------------
1 | """Custom exceptions for FastMCP."""
2 |
3 |
4 | class FastMCPError(Exception):
5 | """Base error for FastMCP."""
6 |
7 |
8 | class ValidationError(FastMCPError):
9 | """Error in validating parameters or return values."""
10 |
11 |
12 | class ResourceError(FastMCPError):
13 | """Error in resource operations."""
14 |
15 |
16 | class ToolError(FastMCPError):
17 | """Error in tool operations."""
18 |
19 |
20 | class InvalidSignature(Exception):
21 | """Invalid signature for use with FastMCP."""
22 |
--------------------------------------------------------------------------------
/src/fastmcp/prompts/__init__.py:
--------------------------------------------------------------------------------
1 | from .base import Prompt
2 | from .manager import PromptManager
3 |
4 | __all__ = ["Prompt", "PromptManager"]
5 |
--------------------------------------------------------------------------------
/src/fastmcp/prompts/base.py:
--------------------------------------------------------------------------------
1 | """Base classes for FastMCP prompts."""
2 |
3 | import json
4 | from typing import Any, Callable, Dict, Literal, Optional, Sequence, Awaitable
5 | import inspect
6 |
7 | from pydantic import BaseModel, Field, TypeAdapter, validate_call
8 | from mcp.types import TextContent, ImageContent, EmbeddedResource
9 | import pydantic_core
10 |
11 | CONTENT_TYPES = TextContent | ImageContent | EmbeddedResource
12 |
13 |
14 | class Message(BaseModel):
15 | """Base class for all prompt messages."""
16 |
17 | role: Literal["user", "assistant"]
18 | content: CONTENT_TYPES
19 |
20 | def __init__(self, content: str | CONTENT_TYPES, **kwargs):
21 | if isinstance(content, str):
22 | content = TextContent(type="text", text=content)
23 | super().__init__(content=content, **kwargs)
24 |
25 |
26 | class UserMessage(Message):
27 | """A message from the user."""
28 |
29 | role: Literal["user"] = "user"
30 |
31 | def __init__(self, content: str | CONTENT_TYPES, **kwargs):
32 | super().__init__(content=content, **kwargs)
33 |
34 |
35 | class AssistantMessage(Message):
36 | """A message from the assistant."""
37 |
38 | role: Literal["assistant"] = "assistant"
39 |
40 | def __init__(self, content: str | CONTENT_TYPES, **kwargs):
41 | super().__init__(content=content, **kwargs)
42 |
43 |
44 | message_validator = TypeAdapter(UserMessage | AssistantMessage)
45 |
46 | SyncPromptResult = (
47 | str | Message | dict[str, Any] | Sequence[str | Message | dict[str, Any]]
48 | )
49 | PromptResult = SyncPromptResult | Awaitable[SyncPromptResult]
50 |
51 |
52 | class PromptArgument(BaseModel):
53 | """An argument that can be passed to a prompt."""
54 |
55 | name: str = Field(description="Name of the argument")
56 | description: str | None = Field(
57 | None, description="Description of what the argument does"
58 | )
59 | required: bool = Field(
60 | default=False, description="Whether the argument is required"
61 | )
62 |
63 |
64 | class Prompt(BaseModel):
65 | """A prompt template that can be rendered with parameters."""
66 |
67 | name: str = Field(description="Name of the prompt")
68 | description: str | None = Field(
69 | None, description="Description of what the prompt does"
70 | )
71 | arguments: list[PromptArgument] | None = Field(
72 | None, description="Arguments that can be passed to the prompt"
73 | )
74 | fn: Callable = Field(exclude=True)
75 |
76 | @classmethod
77 | def from_function(
78 | cls,
79 | fn: Callable[..., PromptResult],
80 | name: Optional[str] = None,
81 | description: Optional[str] = None,
82 | ) -> "Prompt":
83 | """Create a Prompt from a function.
84 |
85 | The function can return:
86 | - A string (converted to a message)
87 | - A Message object
88 | - A dict (converted to a message)
89 | - A sequence of any of the above
90 | """
91 | func_name = name or fn.__name__
92 |
93 | if func_name == "<lambda>":
94 | raise ValueError("You must provide a name for lambda functions")
95 |
96 | # Get schema from TypeAdapter - will fail if function isn't properly typed
97 | parameters = TypeAdapter(fn).json_schema()
98 |
99 | # Convert parameters to PromptArguments
100 | arguments = []
101 | if "properties" in parameters:
102 | for param_name, param in parameters["properties"].items():
103 | required = param_name in parameters.get("required", [])
104 | arguments.append(
105 | PromptArgument(
106 | name=param_name,
107 | description=param.get("description"),
108 | required=required,
109 | )
110 | )
111 |
112 | # ensure the arguments are properly cast
113 | fn = validate_call(fn)
114 |
115 | return cls(
116 | name=func_name,
117 | description=description or fn.__doc__ or "",
118 | arguments=arguments,
119 | fn=fn,
120 | )
121 |
122 | async def render(self, arguments: Optional[Dict[str, Any]] = None) -> list[Message]:
123 | """Render the prompt with arguments."""
124 | # Validate required arguments
125 | if self.arguments:
126 | required = {arg.name for arg in self.arguments if arg.required}
127 | provided = set(arguments or {})
128 | missing = required - provided
129 | if missing:
130 | raise ValueError(f"Missing required arguments: {missing}")
131 |
132 | try:
133 | # Call function and check if result is a coroutine
134 | result = self.fn(**(arguments or {}))
135 | if inspect.iscoroutine(result):
136 | result = await result
137 |
138 | # Validate messages
139 | if not isinstance(result, (list, tuple)):
140 | result = [result]
141 |
142 | # Convert result to messages
143 | messages = []
144 | for msg in result:
145 | try:
146 | if isinstance(msg, Message):
147 | messages.append(msg)
148 | elif isinstance(msg, dict):
149 | msg = message_validator.validate_python(msg)
150 | messages.append(msg)
151 | elif isinstance(msg, str):
152 | messages.append(
153 | UserMessage(content=TextContent(type="text", text=msg))
154 | )
155 | else:
156 | msg = json.dumps(pydantic_core.to_jsonable_python(msg))
157 | messages.append(Message(role="user", content=msg))
158 | except Exception:
159 | raise ValueError(
160 | f"Could not convert prompt result to message: {msg}"
161 | )
162 |
163 | return messages
164 | except Exception as e:
165 | raise ValueError(f"Error rendering prompt {self.name}: {e}")
166 |
--------------------------------------------------------------------------------
/src/fastmcp/prompts/manager.py:
--------------------------------------------------------------------------------
1 | """Prompt management functionality."""
2 |
3 | from typing import Any, Dict, Optional
4 |
5 | from fastmcp.prompts.base import Message, Prompt
6 | from fastmcp.utilities.logging import get_logger
7 |
8 | logger = get_logger(__name__)
9 |
10 |
11 | class PromptManager:
12 | """Manages FastMCP prompts."""
13 |
14 | def __init__(self, warn_on_duplicate_prompts: bool = True):
15 | self._prompts: Dict[str, Prompt] = {}
16 | self.warn_on_duplicate_prompts = warn_on_duplicate_prompts
17 |
18 | def get_prompt(self, name: str) -> Optional[Prompt]:
19 | """Get prompt by name."""
20 | return self._prompts.get(name)
21 |
22 | def list_prompts(self) -> list[Prompt]:
23 | """List all registered prompts."""
24 | return list(self._prompts.values())
25 |
26 | def add_prompt(
27 | self,
28 | prompt: Prompt,
29 | ) -> Prompt:
30 | """Add a prompt to the manager."""
31 |
32 | # Check for duplicates
33 | existing = self._prompts.get(prompt.name)
34 | if existing:
35 | if self.warn_on_duplicate_prompts:
36 | logger.warning(f"Prompt already exists: {prompt.name}")
37 | return existing
38 |
39 | self._prompts[prompt.name] = prompt
40 | return prompt
41 |
42 | async def render_prompt(
43 | self, name: str, arguments: Optional[Dict[str, Any]] = None
44 | ) -> list[Message]:
45 | """Render a prompt by name with arguments."""
46 | prompt = self.get_prompt(name)
47 | if not prompt:
48 | raise ValueError(f"Unknown prompt: {name}")
49 |
50 | return await prompt.render(arguments)
51 |
--------------------------------------------------------------------------------
/src/fastmcp/prompts/prompt_manager.py:
--------------------------------------------------------------------------------
1 | """Prompt management functionality."""
2 |
3 | from typing import Dict, Optional
4 |
5 |
6 | from fastmcp.prompts.base import Prompt
7 | from fastmcp.utilities.logging import get_logger
8 |
9 | logger = get_logger(__name__)
10 |
11 |
12 | class PromptManager:
13 | """Manages FastMCP prompts."""
14 |
15 | def __init__(self, warn_on_duplicate_prompts: bool = True):
16 | self._prompts: Dict[str, Prompt] = {}
17 | self.warn_on_duplicate_prompts = warn_on_duplicate_prompts
18 |
19 | def add_prompt(self, prompt: Prompt) -> Prompt:
20 | """Add a prompt to the manager."""
21 | logger.debug(f"Adding prompt: {prompt.name}")
22 | existing = self._prompts.get(prompt.name)
23 | if existing:
24 | if self.warn_on_duplicate_prompts:
25 | logger.warning(f"Prompt already exists: {prompt.name}")
26 | return existing
27 | self._prompts[prompt.name] = prompt
28 | return prompt
29 |
30 | def get_prompt(self, name: str) -> Optional[Prompt]:
31 | """Get prompt by name."""
32 | return self._prompts.get(name)
33 |
34 | def list_prompts(self) -> list[Prompt]:
35 | """List all registered prompts."""
36 | return list(self._prompts.values())
37 |
--------------------------------------------------------------------------------
/src/fastmcp/py.typed:
--------------------------------------------------------------------------------
https://raw.githubusercontent.com/jlowin/fastmcp/80c328b3dc0b949f010456ee0e85cc5c90e3305f/src/fastmcp/py.typed
--------------------------------------------------------------------------------
/src/fastmcp/resources/__init__.py:
--------------------------------------------------------------------------------
1 | from .base import Resource
2 | from .types import (
3 | TextResource,
4 | BinaryResource,
5 | FunctionResource,
6 | FileResource,
7 | HttpResource,
8 | DirectoryResource,
9 | )
10 | from .templates import ResourceTemplate
11 | from .resource_manager import ResourceManager
12 |
13 | __all__ = [
14 | "Resource",
15 | "TextResource",
16 | "BinaryResource",
17 | "FunctionResource",
18 | "FileResource",
19 | "HttpResource",
20 | "DirectoryResource",
21 | "ResourceTemplate",
22 | "ResourceManager",
23 | ]
24 |
--------------------------------------------------------------------------------
/src/fastmcp/resources/base.py:
--------------------------------------------------------------------------------
1 | """Base classes and interfaces for FastMCP resources."""
2 |
3 | import abc
4 | from typing import Union, Annotated
5 |
6 | from pydantic import (
7 | AnyUrl,
8 | BaseModel,
9 | ConfigDict,
10 | Field,
11 | UrlConstraints,
12 | ValidationInfo,
13 | field_validator,
14 | )
15 |
16 |
17 | class Resource(BaseModel, abc.ABC):
18 | """Base class for all resources."""
19 |
20 | model_config = ConfigDict(validate_default=True)
21 |
22 | uri: Annotated[AnyUrl, UrlConstraints(host_required=False)] = Field(
23 | default=..., description="URI of the resource"
24 | )
25 | name: str | None = Field(description="Name of the resource", default=None)
26 | description: str | None = Field(
27 | description="Description of the resource", default=None
28 | )
29 | mime_type: str = Field(
30 | default="text/plain",
31 | description="MIME type of the resource content",
32 | pattern=r"^[a-zA-Z0-9]+/[a-zA-Z0-9\-+.]+$",
33 | )
34 |
35 | @field_validator("name", mode="before")
36 | @classmethod
37 | def set_default_name(cls, name: str | None, info: ValidationInfo) -> str:
38 | """Set default name from URI if not provided."""
39 | if name:
40 | return name
41 | if uri := info.data.get("uri"):
42 | return str(uri)
43 | raise ValueError("Either name or uri must be provided")
44 |
45 | @abc.abstractmethod
46 | async def read(self) -> Union[str, bytes]:
47 | """Read the resource content."""
48 | pass
49 |
--------------------------------------------------------------------------------
/src/fastmcp/resources/resource_manager.py:
--------------------------------------------------------------------------------
1 | """Resource manager functionality."""
2 |
3 | from typing import Callable, Dict, Optional, Union
4 |
5 | from pydantic import AnyUrl
6 |
7 | from fastmcp.resources.base import Resource
8 | from fastmcp.resources.templates import ResourceTemplate
9 | from fastmcp.utilities.logging import get_logger
10 |
11 | logger = get_logger(__name__)
12 |
13 |
14 | class ResourceManager:
15 | """Manages FastMCP resources."""
16 |
17 | def __init__(self, warn_on_duplicate_resources: bool = True):
18 | self._resources: Dict[str, Resource] = {}
19 | self._templates: Dict[str, ResourceTemplate] = {}
20 | self.warn_on_duplicate_resources = warn_on_duplicate_resources
21 |
22 | def add_resource(self, resource: Resource) -> Resource:
23 | """Add a resource to the manager.
24 |
25 | Args:
26 | resource: A Resource instance to add
27 |
28 | Returns:
29 | The added resource. If a resource with the same URI already exists,
30 | returns the existing resource.
31 | """
32 | logger.debug(
33 | "Adding resource",
34 | extra={
35 | "uri": resource.uri,
36 | "type": type(resource).__name__,
37 | "name": resource.name,
38 | },
39 | )
40 | existing = self._resources.get(str(resource.uri))
41 | if existing:
42 | if self.warn_on_duplicate_resources:
43 | logger.warning(f"Resource already exists: {resource.uri}")
44 | return existing
45 | self._resources[str(resource.uri)] = resource
46 | return resource
47 |
48 | def add_template(
49 | self,
50 | fn: Callable,
51 | uri_template: str,
52 | name: Optional[str] = None,
53 | description: Optional[str] = None,
54 | mime_type: Optional[str] = None,
55 | ) -> ResourceTemplate:
56 | """Add a template from a function."""
57 | template = ResourceTemplate.from_function(
58 | fn,
59 | uri_template=uri_template,
60 | name=name,
61 | description=description,
62 | mime_type=mime_type,
63 | )
64 | self._templates[template.uri_template] = template
65 | return template
66 |
67 | async def get_resource(self, uri: Union[AnyUrl, str]) -> Optional[Resource]:
68 | """Get resource by URI, checking concrete resources first, then templates."""
69 | uri_str = str(uri)
70 | logger.debug("Getting resource", extra={"uri": uri_str})
71 |
72 | # First check concrete resources
73 | if resource := self._resources.get(uri_str):
74 | return resource
75 |
76 | # Then check templates
77 | for template in self._templates.values():
78 | if params := template.matches(uri_str):
79 | try:
80 | return await template.create_resource(uri_str, params)
81 | except Exception as e:
82 | raise ValueError(f"Error creating resource from template: {e}")
83 |
84 | raise ValueError(f"Unknown resource: {uri}")
85 |
86 | def list_resources(self) -> list[Resource]:
87 | """List all registered resources."""
88 | logger.debug("Listing resources", extra={"count": len(self._resources)})
89 | return list(self._resources.values())
90 |
91 | def list_templates(self) -> list[ResourceTemplate]:
92 | """List all registered templates."""
93 | logger.debug("Listing templates", extra={"count": len(self._templates)})
94 | return list(self._templates.values())
95 |
--------------------------------------------------------------------------------
/src/fastmcp/resources/templates.py:
--------------------------------------------------------------------------------
1 | """Resource template functionality."""
2 |
3 | import inspect
4 | import re
5 | from typing import Any, Callable, Dict, Optional
6 |
7 | from pydantic import BaseModel, Field, TypeAdapter, validate_call
8 |
9 | from fastmcp.resources.types import FunctionResource, Resource
10 |
11 |
12 | class ResourceTemplate(BaseModel):
13 | """A template for dynamically creating resources."""
14 |
15 | uri_template: str = Field(
16 | description="URI template with parameters (e.g. weather://{city}/current)"
17 | )
18 | name: str = Field(description="Name of the resource")
19 | description: str | None = Field(description="Description of what the resource does")
20 | mime_type: str = Field(
21 | default="text/plain", description="MIME type of the resource content"
22 | )
23 | fn: Callable = Field(exclude=True)
24 | parameters: dict = Field(description="JSON schema for function parameters")
25 |
26 | @classmethod
27 | def from_function(
28 | cls,
29 | fn: Callable,
30 | uri_template: str,
31 | name: Optional[str] = None,
32 | description: Optional[str] = None,
33 | mime_type: Optional[str] = None,
34 | ) -> "ResourceTemplate":
35 | """Create a template from a function."""
36 | func_name = name or fn.__name__
37 | if func_name == "<lambda>":
38 | raise ValueError("You must provide a name for lambda functions")
39 |
40 | # Get schema from TypeAdapter - will fail if function isn't properly typed
41 | parameters = TypeAdapter(fn).json_schema()
42 |
43 | # ensure the arguments are properly cast
44 | fn = validate_call(fn)
45 |
46 | return cls(
47 | uri_template=uri_template,
48 | name=func_name,
49 | description=description or fn.__doc__ or "",
50 | mime_type=mime_type or "text/plain",
51 | fn=fn,
52 | parameters=parameters,
53 | )
54 |
55 | def matches(self, uri: str) -> Optional[Dict[str, Any]]:
56 | """Check if URI matches template and extract parameters."""
57 | # Convert template to regex pattern
58 | pattern = self.uri_template.replace("{", "(?P<").replace("}", ">[^/]+)")
59 | match = re.match(f"^{pattern}$", uri)
60 | if match:
61 | return match.groupdict()
62 | return None
63 |
64 | async def create_resource(self, uri: str, params: Dict[str, Any]) -> Resource:
65 | """Create a resource from the template with the given parameters."""
66 | try:
67 | # Call function and check if result is a coroutine
68 | result = self.fn(**params)
69 | if inspect.iscoroutine(result):
70 | result = await result
71 |
72 | return FunctionResource(
73 | uri=uri, # type: ignore
74 | name=self.name,
75 | description=self.description,
76 | mime_type=self.mime_type,
77 | fn=lambda: result, # Capture result in closure
78 | )
79 | except Exception as e:
80 | raise ValueError(f"Error creating resource from template: {e}")
81 |
--------------------------------------------------------------------------------
/src/fastmcp/resources/types.py:
--------------------------------------------------------------------------------
1 | """Concrete resource implementations."""
2 |
3 | import asyncio
4 | import json
5 | from pathlib import Path
6 | from typing import Any, Callable, Union
7 |
8 | import httpx
9 | import pydantic.json
10 | import pydantic_core
11 | from pydantic import Field, ValidationInfo
12 |
13 | from fastmcp.resources.base import Resource
14 |
15 |
16 | class TextResource(Resource):
17 | """A resource that reads from a string."""
18 |
19 | text: str = Field(description="Text content of the resource")
20 |
21 | async def read(self) -> str:
22 | """Read the text content."""
23 | return self.text
24 |
25 |
26 | class BinaryResource(Resource):
27 | """A resource that reads from bytes."""
28 |
29 | data: bytes = Field(description="Binary content of the resource")
30 |
31 | async def read(self) -> bytes:
32 | """Read the binary content."""
33 | return self.data
34 |
35 |
36 | class FunctionResource(Resource):
37 | """A resource that defers data loading by wrapping a function.
38 |
39 | The function is only called when the resource is read, allowing for lazy loading
40 | of potentially expensive data. This is particularly useful when listing resources,
41 | as the function won't be called until the resource is actually accessed.
42 |
43 | The function can return:
44 | - str for text content (default)
45 | - bytes for binary content
46 | - other types will be converted to JSON
47 | """
48 |
49 | fn: Callable[[], Any] = Field(exclude=True)
50 |
51 | async def read(self) -> Union[str, bytes]:
52 | """Read the resource by calling the wrapped function."""
53 | try:
54 | result = self.fn()
55 | if isinstance(result, Resource):
56 | return await result.read()
57 | if isinstance(result, bytes):
58 | return result
59 | if isinstance(result, str):
60 | return result
61 | try:
62 | return json.dumps(pydantic_core.to_jsonable_python(result))
63 | except (TypeError, pydantic_core.PydanticSerializationError):
64 | # If JSON serialization fails, try str()
65 | return str(result)
66 | except Exception as e:
67 | raise ValueError(f"Error reading resource {self.uri}: {e}")
68 |
69 |
70 | class FileResource(Resource):
71 | """A resource that reads from a file.
72 |
73 | Set is_binary=True to read file as binary data instead of text.
74 | """
75 |
76 | path: Path = Field(description="Path to the file")
77 | is_binary: bool = Field(
78 | default=False,
79 | description="Whether to read the file as binary data",
80 | )
81 | mime_type: str = Field(
82 | default="text/plain",
83 | description="MIME type of the resource content",
84 | )
85 |
86 | @pydantic.field_validator("path")
87 | @classmethod
88 | def validate_absolute_path(cls, path: Path) -> Path:
89 | """Ensure path is absolute."""
90 | if not path.is_absolute():
91 | raise ValueError("Path must be absolute")
92 | return path
93 |
94 | @pydantic.field_validator("is_binary")
95 | @classmethod
96 | def set_binary_from_mime_type(cls, is_binary: bool, info: ValidationInfo) -> bool:
97 | """Set is_binary based on mime_type if not explicitly set."""
98 | if is_binary:
99 | return True
100 | mime_type = info.data.get("mime_type", "text/plain")
101 | return not mime_type.startswith("text/")
102 |
103 | async def read(self) -> Union[str, bytes]:
104 | """Read the file content."""
105 | try:
106 | if self.is_binary:
107 | return await asyncio.to_thread(self.path.read_bytes)
108 | return await asyncio.to_thread(self.path.read_text)
109 | except Exception as e:
110 | raise ValueError(f"Error reading file {self.path}: {e}")
111 |
112 |
113 | class HttpResource(Resource):
114 | """A resource that reads from an HTTP endpoint."""
115 |
116 | url: str = Field(description="URL to fetch content from")
117 | mime_type: str | None = Field(
118 | default="application/json", description="MIME type of the resource content"
119 | )
120 |
121 | async def read(self) -> Union[str, bytes]:
122 | """Read the HTTP content."""
123 | async with httpx.AsyncClient() as client:
124 | response = await client.get(self.url)
125 | response.raise_for_status()
126 | return response.text
127 |
128 |
129 | class DirectoryResource(Resource):
130 | """A resource that lists files in a directory."""
131 |
132 | path: Path = Field(description="Path to the directory")
133 | recursive: bool = Field(
134 | default=False, description="Whether to list files recursively"
135 | )
136 | pattern: str | None = Field(
137 | default=None, description="Optional glob pattern to filter files"
138 | )
139 | mime_type: str | None = Field(
140 | default="application/json", description="MIME type of the resource content"
141 | )
142 |
143 | @pydantic.field_validator("path")
144 | @classmethod
145 | def validate_absolute_path(cls, path: Path) -> Path:
146 | """Ensure path is absolute."""
147 | if not path.is_absolute():
148 | raise ValueError("Path must be absolute")
149 | return path
150 |
151 | def list_files(self) -> list[Path]:
152 | """List files in the directory."""
153 | if not self.path.exists():
154 | raise FileNotFoundError(f"Directory not found: {self.path}")
155 | if not self.path.is_dir():
156 | raise NotADirectoryError(f"Not a directory: {self.path}")
157 |
158 | try:
159 | if self.pattern:
160 | return (
161 | list(self.path.glob(self.pattern))
162 | if not self.recursive
163 | else list(self.path.rglob(self.pattern))
164 | )
165 | return (
166 | list(self.path.glob("*"))
167 | if not self.recursive
168 | else list(self.path.rglob("*"))
169 | )
170 | except Exception as e:
171 | raise ValueError(f"Error listing directory {self.path}: {e}")
172 |
173 | async def read(self) -> str: # Always returns JSON string
174 | """Read the directory listing."""
175 | try:
176 | files = await asyncio.to_thread(self.list_files)
177 | file_list = [str(f.relative_to(self.path)) for f in files if f.is_file()]
178 | return json.dumps({"files": file_list}, indent=2)
179 | except Exception as e:
180 | raise ValueError(f"Error reading directory {self.path}: {e}")
181 |
--------------------------------------------------------------------------------
/src/fastmcp/server.py:
--------------------------------------------------------------------------------
1 | """FastMCP - A more ergonomic interface for MCP servers."""
2 |
3 | import asyncio
4 | import functools
5 | import inspect
6 | import json
7 | import re
8 | from itertools import chain
9 | from typing import Any, Callable, Dict, Literal, Sequence, TypeVar, ParamSpec
10 |
11 | import pydantic_core
12 | from pydantic import Field
13 | import uvicorn
14 | from mcp.server import Server as MCPServer
15 | from mcp.server.sse import SseServerTransport
16 | from mcp.server.stdio import stdio_server
17 | from mcp.shared.context import RequestContext
18 | from mcp.types import (
19 | EmbeddedResource,
20 | GetPromptResult,
21 | ImageContent,
22 | TextContent,
23 | )
24 | from mcp.types import (
25 | Prompt as MCPPrompt,
26 | PromptArgument as MCPPromptArgument,
27 | )
28 | from mcp.types import (
29 | Resource as MCPResource,
30 | )
31 | from mcp.types import (
32 | ResourceTemplate as MCPResourceTemplate,
33 | )
34 | from mcp.types import (
35 | Tool as MCPTool,
36 | )
37 | from pydantic import BaseModel
38 | from pydantic.networks import AnyUrl
39 | from pydantic_settings import BaseSettings, SettingsConfigDict
40 |
41 | from fastmcp.exceptions import ResourceError
42 | from fastmcp.prompts import Prompt, PromptManager
43 | from fastmcp.prompts.base import PromptResult
44 | from fastmcp.resources import FunctionResource, Resource, ResourceManager
45 | from fastmcp.tools import ToolManager
46 | from fastmcp.utilities.logging import configure_logging, get_logger
47 | from fastmcp.utilities.types import Image
48 |
49 | logger = get_logger(__name__)
50 |
51 | P = ParamSpec("P")
52 | R = TypeVar("R")
53 | R_PromptResult = TypeVar("R_PromptResult", bound=PromptResult)
54 |
55 |
56 | class Settings(BaseSettings):
57 | """FastMCP server settings.
58 |
59 | All settings can be configured via environment variables with the prefix FASTMCP_.
60 | For example, FASTMCP_DEBUG=true will set debug=True.
61 | """
62 |
63 | model_config: SettingsConfigDict = SettingsConfigDict(
64 | env_prefix="FASTMCP_",
65 | env_file=".env",
66 | extra="ignore",
67 | )
68 |
69 | # Server settings
70 | debug: bool = False
71 | log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO"
72 |
73 | # HTTP settings
74 | host: str = "0.0.0.0"
75 | port: int = 8000
76 |
77 | # resource settings
78 | warn_on_duplicate_resources: bool = True
79 |
80 | # tool settings
81 | warn_on_duplicate_tools: bool = True
82 |
83 | # prompt settings
84 | warn_on_duplicate_prompts: bool = True
85 |
86 | dependencies: list[str] = Field(
87 | default_factory=list,
88 | description="List of dependencies to install in the server environment",
89 | )
90 |
91 |
92 | class FastMCP:
93 | def __init__(self, name: str | None = None, **settings: Any):
94 | self.settings = Settings(**settings)
95 | self._mcp_server = MCPServer(name=name or "FastMCP")
96 | self._tool_manager = ToolManager(
97 | warn_on_duplicate_tools=self.settings.warn_on_duplicate_tools
98 | )
99 | self._resource_manager = ResourceManager(
100 | warn_on_duplicate_resources=self.settings.warn_on_duplicate_resources
101 | )
102 | self._prompt_manager = PromptManager(
103 | warn_on_duplicate_prompts=self.settings.warn_on_duplicate_prompts
104 | )
105 | self.dependencies = self.settings.dependencies
106 |
107 | # Set up MCP protocol handlers
108 | self._setup_handlers()
109 |
110 | # Configure logging
111 | configure_logging(self.settings.log_level)
112 |
113 | @property
114 | def name(self) -> str:
115 | return self._mcp_server.name
116 |
117 | def run(self, transport: Literal["stdio", "sse"] = "stdio") -> None:
118 | """Run the FastMCP server. Note this is a synchronous function.
119 |
120 | Args:
121 | transport: Transport protocol to use ("stdio" or "sse")
122 | """
123 | TRANSPORTS = Literal["stdio", "sse"]
124 | if transport not in TRANSPORTS.__args__: # type: ignore
125 | raise ValueError(f"Unknown transport: {transport}")
126 |
127 | if transport == "stdio":
128 | asyncio.run(self.run_stdio_async())
129 | else: # transport == "sse"
130 | asyncio.run(self.run_sse_async())
131 |
132 | def _setup_handlers(self) -> None:
133 | """Set up core MCP protocol handlers."""
134 | self._mcp_server.list_tools()(self.list_tools)
135 | self._mcp_server.call_tool()(self.call_tool)
136 | self._mcp_server.list_resources()(self.list_resources)
137 | self._mcp_server.read_resource()(self.read_resource)
138 | self._mcp_server.list_prompts()(self.list_prompts)
139 | self._mcp_server.get_prompt()(self.get_prompt)
140 | # TODO: This has not been added to MCP yet, see https://github.com/jlowin/fastmcp/issues/10
141 | # self._mcp_server.list_resource_templates()(self.list_resource_templates)
142 |
143 | async def list_tools(self) -> list[MCPTool]:
144 | """List all available tools."""
145 | tools = self._tool_manager.list_tools()
146 | return [
147 | MCPTool(
148 | name=info.name,
149 | description=info.description,
150 | inputSchema=info.parameters,
151 | )
152 | for info in tools
153 | ]
154 |
155 | def get_context(self) -> "Context":
156 | """
157 | Returns a Context object. Note that the context will only be valid
158 | during a request; outside a request, most methods will error.
159 | """
160 | try:
161 | request_context = self._mcp_server.request_context
162 | except LookupError:
163 | request_context = None
164 | return Context(request_context=request_context, fastmcp=self)
165 |
166 | async def call_tool(
167 | self, name: str, arguments: dict
168 | ) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
169 | """Call a tool by name with arguments."""
170 | context = self.get_context()
171 | result = await self._tool_manager.call_tool(name, arguments, context=context)
172 | converted_result = _convert_to_content(result)
173 | return converted_result
174 |
175 | async def list_resources(self) -> list[MCPResource]:
176 | """List all available resources."""
177 |
178 | resources = self._resource_manager.list_resources()
179 | return [
180 | MCPResource(
181 | uri=resource.uri,
182 | name=resource.name or "",
183 | description=resource.description,
184 | mimeType=resource.mime_type,
185 | )
186 | for resource in resources
187 | ]
188 |
189 | async def list_resource_templates(self) -> list[MCPResourceTemplate]:
190 | templates = self._resource_manager.list_templates()
191 | return [
192 | MCPResourceTemplate(
193 | uriTemplate=template.uri_template,
194 | name=template.name,
195 | description=template.description,
196 | )
197 | for template in templates
198 | ]
199 |
200 | async def read_resource(self, uri: AnyUrl | str) -> str | bytes:
201 | """Read a resource by URI."""
202 | resource = await self._resource_manager.get_resource(uri)
203 | if not resource:
204 | raise ResourceError(f"Unknown resource: {uri}")
205 |
206 | try:
207 | return await resource.read()
208 | except Exception as e:
209 | logger.error(f"Error reading resource {uri}: {e}")
210 | raise ResourceError(str(e))
211 |
212 | def add_tool(
213 | self,
214 | fn: Callable,
215 | name: str | None = None,
216 | description: str | None = None,
217 | ) -> None:
218 | """Add a tool to the server.
219 |
220 | The tool function can optionally request a Context object by adding a parameter
221 | with the Context type annotation. See the @tool decorator for examples.
222 |
223 | Args:
224 | fn: The function to register as a tool
225 | name: Optional name for the tool (defaults to function name)
226 | description: Optional description of what the tool does
227 | """
228 | self._tool_manager.add_tool(fn, name=name, description=description)
229 |
230 | def tool(
231 | self, name: str | None = None, description: str | None = None
232 | ) -> Callable[[Callable[P, R]], Callable[P, R]]:
233 | """Decorator to register a tool.
234 |
235 | Tools can optionally request a Context object by adding a parameter with the Context type annotation.
236 | The context provides access to MCP capabilities like logging, progress reporting, and resource access.
237 |
238 | Args:
239 | name: Optional name for the tool (defaults to function name)
240 | description: Optional description of what the tool does
241 |
242 | Example:
243 | @server.tool()
244 | def my_tool(x: int) -> str:
245 | return str(x)
246 |
247 | @server.tool()
248 | def tool_with_context(x: int, ctx: Context) -> str:
249 | ctx.info(f"Processing {x}")
250 | return str(x)
251 |
252 | @server.tool()
253 | async def async_tool(x: int, context: Context) -> str:
254 | await context.report_progress(50, 100)
255 | return str(x)
256 | """
257 | # Check if user passed function directly instead of calling decorator
258 | if callable(name):
259 | raise TypeError(
260 | "The @tool decorator was used incorrectly. "
261 | "Did you forget to call it? Use @tool() instead of @tool"
262 | )
263 |
264 | def decorator(fn: Callable[P, R]) -> Callable[P, R]:
265 | self.add_tool(fn, name=name, description=description)
266 | return fn
267 |
268 | return decorator
269 |
270 | def add_resource(self, resource: Resource) -> None:
271 | """Add a resource to the server.
272 |
273 | Args:
274 | resource: A Resource instance to add
275 | """
276 | self._resource_manager.add_resource(resource)
277 |
278 | def resource(
279 | self,
280 | uri: str,
281 | *,
282 | name: str | None = None,
283 | description: str | None = None,
284 | mime_type: str | None = None,
285 | ) -> Callable[[Callable[P, R]], Callable[P, R]]:
286 | """Decorator to register a function as a resource.
287 |
288 | The function will be called when the resource is read to generate its content.
289 | The function can return:
290 | - str for text content
291 | - bytes for binary content
292 | - other types will be converted to JSON
293 |
294 | If the URI contains parameters (e.g. "resource://{param}") or the function
295 | has parameters, it will be registered as a template resource.
296 |
297 | Args:
298 | uri: URI for the resource (e.g. "resource://my-resource" or "resource://{param}")
299 | name: Optional name for the resource
300 | description: Optional description of the resource
301 | mime_type: Optional MIME type for the resource
302 |
303 | Example:
304 | @server.resource("resource://my-resource")
305 | def get_data() -> str:
306 | return "Hello, world!"
307 |
308 | @server.resource("resource://{city}/weather")
309 | def get_weather(city: str) -> str:
310 | return f"Weather for {city}"
311 | """
312 | # Check if user passed function directly instead of calling decorator
313 | if callable(uri):
314 | raise TypeError(
315 | "The @resource decorator was used incorrectly. "
316 | "Did you forget to call it? Use @resource('uri') instead of @resource"
317 | )
318 |
319 | def decorator(fn: Callable[P, R]) -> Callable[P, R]:
320 | @functools.wraps(fn)
321 | def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
322 | return fn(*args, **kwargs)
323 |
324 | # Check if this should be a template
325 | has_uri_params = "{" in uri and "}" in uri
326 | has_func_params = bool(inspect.signature(fn).parameters)
327 |
328 | if has_uri_params or has_func_params:
329 | # Validate that URI params match function params
330 | uri_params = set(re.findall(r"{(\w+)}", uri))
331 | func_params = set(inspect.signature(fn).parameters.keys())
332 |
333 | if uri_params != func_params:
334 | raise ValueError(
335 | f"Mismatch between URI parameters {uri_params} "
336 | f"and function parameters {func_params}"
337 | )
338 |
339 | # Register as template
340 | self._resource_manager.add_template(
341 | wrapper,
342 | uri_template=uri,
343 | name=name,
344 | description=description,
345 | mime_type=mime_type or "text/plain",
346 | )
347 | else:
348 | # Register as regular resource
349 | resource = FunctionResource(
350 | uri=AnyUrl(uri),
351 | name=name,
352 | description=description,
353 | mime_type=mime_type or "text/plain",
354 | fn=wrapper,
355 | )
356 | self.add_resource(resource)
357 | return wrapper
358 |
359 | return decorator
360 |
361 | def add_prompt(self, prompt: Prompt) -> None:
362 | """Add a prompt to the server.
363 |
364 | Args:
365 | prompt: A Prompt instance to add
366 | """
367 | self._prompt_manager.add_prompt(prompt)
368 |
369 | def prompt(
370 | self, name: str | None = None, description: str | None = None
371 | ) -> Callable[[Callable[P, R_PromptResult]], Callable[P, R_PromptResult]]:
372 | """Decorator to register a prompt.
373 |
374 | Args:
375 | name: Optional name for the prompt (defaults to function name)
376 | description: Optional description of what the prompt does
377 |
378 | Example:
379 | @server.prompt()
380 | def analyze_table(table_name: str) -> list[Message]:
381 | schema = read_table_schema(table_name)
382 | return [
383 | {
384 | "role": "user",
385 | "content": f"Analyze this schema:\n{schema}"
386 | }
387 | ]
388 |
389 | @server.prompt()
390 | async def analyze_file(path: str) -> list[Message]:
391 | content = await read_file(path)
392 | return [
393 | {
394 | "role": "user",
395 | "content": {
396 | "type": "resource",
397 | "resource": {
398 | "uri": f"file://{path}",
399 | "text": content
400 | }
401 | }
402 | }
403 | ]
404 | """
405 | # Check if user passed function directly instead of calling decorator
406 | if callable(name):
407 | raise TypeError(
408 | "The @prompt decorator was used incorrectly. "
409 | "Did you forget to call it? Use @prompt() instead of @prompt"
410 | )
411 |
412 | def decorator(func: Callable[P, R_PromptResult]) -> Callable[P, R_PromptResult]:
413 | prompt = Prompt.from_function(func, name=name, description=description)
414 | self.add_prompt(prompt)
415 | return func
416 |
417 | return decorator
418 |
419 | async def run_stdio_async(self) -> None:
420 | """Run the server using stdio transport."""
421 | async with stdio_server() as (read_stream, write_stream):
422 | await self._mcp_server.run(
423 | read_stream,
424 | write_stream,
425 | self._mcp_server.create_initialization_options(),
426 | )
427 |
428 | async def run_sse_async(self) -> None:
429 | """Run the server using SSE transport."""
430 | from starlette.applications import Starlette
431 | from starlette.routing import Route
432 |
433 | sse = SseServerTransport("/messages")
434 |
435 | async def handle_sse(request):
436 | async with sse.connect_sse(
437 | request.scope, request.receive, request._send
438 | ) as streams:
439 | await self._mcp_server.run(
440 | streams[0],
441 | streams[1],
442 | self._mcp_server.create_initialization_options(),
443 | )
444 |
445 | async def handle_messages(request):
446 | await sse.handle_post_message(request.scope, request.receive, request._send)
447 |
448 | starlette_app = Starlette(
449 | debug=self.settings.debug,
450 | routes=[
451 | Route("/sse", endpoint=handle_sse),
452 | Route("/messages", endpoint=handle_messages, methods=["POST"]),
453 | ],
454 | )
455 |
456 | config = uvicorn.Config(
457 | starlette_app,
458 | host=self.settings.host,
459 | port=self.settings.port,
460 | log_level=self.settings.log_level.lower(),
461 | )
462 | server = uvicorn.Server(config)
463 | await server.serve()
464 |
465 | async def list_prompts(self) -> list[MCPPrompt]:
466 | """List all available prompts."""
467 | prompts = self._prompt_manager.list_prompts()
468 | return [
469 | MCPPrompt(
470 | name=prompt.name,
471 | description=prompt.description,
472 | arguments=[
473 | MCPPromptArgument(
474 | name=arg.name,
475 | description=arg.description,
476 | required=arg.required,
477 | )
478 | for arg in (prompt.arguments or [])
479 | ],
480 | )
481 | for prompt in prompts
482 | ]
483 |
484 | async def get_prompt(
485 | self, name: str, arguments: Dict[str, Any] | None = None
486 | ) -> GetPromptResult:
487 | """Get a prompt by name with arguments."""
488 | try:
489 | messages = await self._prompt_manager.render_prompt(name, arguments)
490 |
491 | return GetPromptResult(messages=pydantic_core.to_jsonable_python(messages))
492 | except Exception as e:
493 | logger.error(f"Error getting prompt {name}: {e}")
494 | raise ValueError(str(e))
495 |
496 |
497 | def _convert_to_content(
498 | result: Any,
499 | ) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
500 | """Convert a result to a sequence of content objects."""
501 | if result is None:
502 | return []
503 |
504 | if isinstance(result, (TextContent, ImageContent, EmbeddedResource)):
505 | return [result]
506 |
507 | if isinstance(result, Image):
508 | return [result.to_image_content()]
509 |
510 | if isinstance(result, (list, tuple)):
511 | return list(chain.from_iterable(_convert_to_content(item) for item in result))
512 |
513 | if not isinstance(result, str):
514 | try:
515 | result = json.dumps(pydantic_core.to_jsonable_python(result))
516 | except Exception:
517 | result = str(result)
518 |
519 | return [TextContent(type="text", text=result)]
520 |
521 |
522 | class Context(BaseModel):
523 | """Context object providing access to MCP capabilities.
524 |
525 | This provides a cleaner interface to MCP's RequestContext functionality.
526 | It gets injected into tool and resource functions that request it via type hints.
527 |
528 | To use context in a tool function, add a parameter with the Context type annotation:
529 |
530 | ```python
531 | @server.tool()
532 | def my_tool(x: int, ctx: Context) -> str:
533 | # Log messages to the client
534 | ctx.info(f"Processing {x}")
535 | ctx.debug("Debug info")
536 | ctx.warning("Warning message")
537 | ctx.error("Error message")
538 |
539 | # Report progress
540 | ctx.report_progress(50, 100)
541 |
542 | # Access resources
543 | data = ctx.read_resource("resource://data")
544 |
545 | # Get request info
546 | request_id = ctx.request_id
547 | client_id = ctx.client_id
548 |
549 | return str(x)
550 | ```
551 |
552 | The context parameter name can be anything as long as it's annotated with Context.
553 | The context is optional - tools that don't need it can omit the parameter.
554 | """
555 |
556 | _request_context: RequestContext | None
557 | _fastmcp: FastMCP | None
558 |
559 | def __init__(
560 | self,
561 | *,
562 | request_context: RequestContext | None = None,
563 | fastmcp: FastMCP | None = None,
564 | **kwargs: Any,
565 | ):
566 | super().__init__(**kwargs)
567 | self._request_context = request_context
568 | self._fastmcp = fastmcp
569 |
570 | @property
571 | def fastmcp(self) -> FastMCP:
572 | """Access to the FastMCP server."""
573 | if self._fastmcp is None:
574 | raise ValueError("Context is not available outside of a request")
575 | return self._fastmcp
576 |
577 | @property
578 | def request_context(self) -> RequestContext:
579 | """Access to the underlying request context."""
580 | if self._request_context is None:
581 | raise ValueError("Context is not available outside of a request")
582 | return self._request_context
583 |
584 | async def report_progress(
585 | self, progress: float, total: float | None = None
586 | ) -> None:
587 | """Report progress for the current operation.
588 |
589 | Args:
590 | progress: Current progress value e.g. 24
591 | total: Optional total value e.g. 100
592 | """
593 |
594 | progress_token = (
595 | self.request_context.meta.progressToken
596 | if self.request_context.meta
597 | else None
598 | )
599 |
600 | if not progress_token:
601 | return
602 |
603 | await self.request_context.session.send_progress_notification(
604 | progress_token=progress_token, progress=progress, total=total
605 | )
606 |
607 | async def read_resource(self, uri: str | AnyUrl) -> str | bytes:
608 | """Read a resource by URI.
609 |
610 | Args:
611 | uri: Resource URI to read
612 |
613 | Returns:
614 | The resource content as either text or bytes
615 | """
616 | assert (
617 | self._fastmcp is not None
618 | ), "Context is not available outside of a request"
619 | return await self._fastmcp.read_resource(uri)
620 |
621 | def log(
622 | self,
623 | level: Literal["debug", "info", "warning", "error"],
624 | message: str,
625 | *,
626 | logger_name: str | None = None,
627 | ) -> None:
628 | """Send a log message to the client.
629 |
630 | Args:
631 | level: Log level (debug, info, warning, error)
632 | message: Log message
633 | logger_name: Optional logger name
634 | **extra: Additional structured data to include
635 | """
636 | self.request_context.session.send_log_message(
637 | level=level, data=message, logger=logger_name
638 | )
639 |
640 | @property
641 | def client_id(self) -> str | None:
642 | """Get the client ID if available."""
643 | return (
644 | getattr(self.request_context.meta, "client_id", None)
645 | if self.request_context.meta
646 | else None
647 | )
648 |
649 | @property
650 | def request_id(self) -> str:
651 | """Get the unique ID for this request."""
652 | return str(self.request_context.request_id)
653 |
654 | @property
655 | def session(self):
656 | """Access to the underlying session for advanced usage."""
657 | return self.request_context.session
658 |
659 | # Convenience methods for common log levels
660 | def debug(self, message: str, **extra: Any) -> None:
661 | """Send a debug log message."""
662 | self.log("debug", message, **extra)
663 |
664 | def info(self, message: str, **extra: Any) -> None:
665 | """Send an info log message."""
666 | self.log("info", message, **extra)
667 |
668 | def warning(self, message: str, **extra: Any) -> None:
669 | """Send a warning log message."""
670 | self.log("warning", message, **extra)
671 |
672 | def error(self, message: str, **extra: Any) -> None:
673 | """Send an error log message."""
674 | self.log("error", message, **extra)
675 |
--------------------------------------------------------------------------------
/src/fastmcp/tools/__init__.py:
--------------------------------------------------------------------------------
1 | from .base import Tool
2 | from .tool_manager import ToolManager
3 |
4 | __all__ = ["Tool", "ToolManager"]
5 |
--------------------------------------------------------------------------------
/src/fastmcp/tools/base.py:
--------------------------------------------------------------------------------
1 | import fastmcp
2 | from fastmcp.exceptions import ToolError
3 |
4 | from fastmcp.utilities.func_metadata import func_metadata, FuncMetadata
5 | from pydantic import BaseModel, Field
6 |
7 |
8 | import inspect
9 | from typing import TYPE_CHECKING, Any, Callable, Optional
10 |
11 | if TYPE_CHECKING:
12 | from fastmcp.server import Context
13 |
14 |
15 | class Tool(BaseModel):
16 | """Internal tool registration info."""
17 |
18 | fn: Callable = Field(exclude=True)
19 | name: str = Field(description="Name of the tool")
20 | description: str = Field(description="Description of what the tool does")
21 | parameters: dict = Field(description="JSON schema for tool parameters")
22 | fn_metadata: FuncMetadata = Field(
23 | description="Metadata about the function including a pydantic model for tool arguments"
24 | )
25 | is_async: bool = Field(description="Whether the tool is async")
26 | context_kwarg: Optional[str] = Field(
27 | None, description="Name of the kwarg that should receive context"
28 | )
29 |
30 | @classmethod
31 | def from_function(
32 | cls,
33 | fn: Callable,
34 | name: Optional[str] = None,
35 | description: Optional[str] = None,
36 | context_kwarg: Optional[str] = None,
37 | ) -> "Tool":
38 | """Create a Tool from a function."""
39 | func_name = name or fn.__name__
40 |
41 | if func_name == "<lambda>":
42 | raise ValueError("You must provide a name for lambda functions")
43 |
44 | func_doc = description or fn.__doc__ or ""
45 | is_async = inspect.iscoroutinefunction(fn)
46 |
47 | # Find context parameter if it exists
48 | if context_kwarg is None:
49 | sig = inspect.signature(fn)
50 | for param_name, param in sig.parameters.items():
51 | if param.annotation is fastmcp.Context:
52 | context_kwarg = param_name
53 | break
54 |
55 | func_arg_metadata = func_metadata(
56 | fn,
57 | skip_names=[context_kwarg] if context_kwarg is not None else [],
58 | )
59 | parameters = func_arg_metadata.arg_model.model_json_schema()
60 |
61 | return cls(
62 | fn=fn,
63 | name=func_name,
64 | description=func_doc,
65 | parameters=parameters,
66 | fn_metadata=func_arg_metadata,
67 | is_async=is_async,
68 | context_kwarg=context_kwarg,
69 | )
70 |
71 | async def run(self, arguments: dict, context: Optional["Context"] = None) -> Any:
72 | """Run the tool with arguments."""
73 | try:
74 | return await self.fn_metadata.call_fn_with_arg_validation(
75 | self.fn,
76 | self.is_async,
77 | arguments,
78 | {self.context_kwarg: context}
79 | if self.context_kwarg is not None
80 | else None,
81 | )
82 | except Exception as e:
83 | raise ToolError(f"Error executing tool {self.name}: {e}") from e
84 |
--------------------------------------------------------------------------------
/src/fastmcp/tools/tool_manager.py:
--------------------------------------------------------------------------------
1 | from fastmcp.exceptions import ToolError
2 |
3 | from fastmcp.tools.base import Tool
4 |
5 |
6 | from typing import Any, Callable, Dict, Optional, TYPE_CHECKING
7 |
8 | from fastmcp.utilities.logging import get_logger
9 |
10 | if TYPE_CHECKING:
11 | from fastmcp.server import Context
12 |
13 | logger = get_logger(__name__)
14 |
15 |
16 | class ToolManager:
17 | """Manages FastMCP tools."""
18 |
19 | def __init__(self, warn_on_duplicate_tools: bool = True):
20 | self._tools: Dict[str, Tool] = {}
21 | self.warn_on_duplicate_tools = warn_on_duplicate_tools
22 |
23 | def get_tool(self, name: str) -> Optional[Tool]:
24 | """Get tool by name."""
25 | return self._tools.get(name)
26 |
27 | def list_tools(self) -> list[Tool]:
28 | """List all registered tools."""
29 | return list(self._tools.values())
30 |
31 | def add_tool(
32 | self,
33 | fn: Callable,
34 | name: Optional[str] = None,
35 | description: Optional[str] = None,
36 | ) -> Tool:
37 | """Add a tool to the server."""
38 | tool = Tool.from_function(fn, name=name, description=description)
39 | existing = self._tools.get(tool.name)
40 | if existing:
41 | if self.warn_on_duplicate_tools:
42 | logger.warning(f"Tool already exists: {tool.name}")
43 | return existing
44 | self._tools[tool.name] = tool
45 | return tool
46 |
47 | async def call_tool(
48 | self, name: str, arguments: dict, context: Optional["Context"] = None
49 | ) -> Any:
50 | """Call a tool by name with arguments."""
51 | tool = self.get_tool(name)
52 | if not tool:
53 | raise ToolError(f"Unknown tool: {name}")
54 |
55 | return await tool.run(arguments, context=context)
56 |
--------------------------------------------------------------------------------
/src/fastmcp/utilities/__init__.py:
--------------------------------------------------------------------------------
1 | """FastMCP utility modules."""
2 |
--------------------------------------------------------------------------------
/src/fastmcp/utilities/func_metadata.py:
--------------------------------------------------------------------------------
1 | import inspect
2 | from collections.abc import Callable, Sequence, Awaitable
3 | from typing import (
4 | Annotated,
5 | Any,
6 | Dict,
7 | ForwardRef,
8 | )
9 | from pydantic import Field
10 | from fastmcp.exceptions import InvalidSignature
11 | from pydantic._internal._typing_extra import eval_type_lenient
12 | import json
13 | from pydantic import BaseModel
14 | from pydantic.fields import FieldInfo
15 | from pydantic import ConfigDict, create_model
16 | from pydantic import WithJsonSchema
17 | from pydantic_core import PydanticUndefined
18 | from fastmcp.utilities.logging import get_logger
19 |
20 |
21 | logger = get_logger(__name__)
22 |
23 |
24 | class ArgModelBase(BaseModel):
25 | """A model representing the arguments to a function."""
26 |
27 | def model_dump_one_level(self) -> dict[str, Any]:
28 | """Return a dict of the model's fields, one level deep.
29 |
30 | That is, sub-models etc are not dumped - they are kept as pydantic models.
31 | """
32 | kwargs: dict[str, Any] = {}
33 | for field_name in self.model_fields.keys():
34 | kwargs[field_name] = getattr(self, field_name)
35 | return kwargs
36 |
37 | model_config = ConfigDict(
38 | arbitrary_types_allowed=True,
39 | )
40 |
41 |
42 | class FuncMetadata(BaseModel):
43 | arg_model: Annotated[type[ArgModelBase], WithJsonSchema(None)]
44 | # We can add things in the future like
45 | # - Maybe some args are excluded from attempting to parse from JSON
46 | # - Maybe some args are special (like context) for dependency injection
47 |
48 | async def call_fn_with_arg_validation(
49 | self,
50 | fn: Callable[..., Any] | Awaitable[Any],
51 | fn_is_async: bool,
52 | arguments_to_validate: dict[str, Any],
53 | arguments_to_pass_directly: dict[str, Any] | None,
54 | ) -> Any:
55 | """Call the given function with arguments validated and injected.
56 |
57 | Arguments are first attempted to be parsed from JSON, then validated against
58 | the argument model, before being passed to the function.
59 | """
60 | arguments_pre_parsed = self.pre_parse_json(arguments_to_validate)
61 | arguments_parsed_model = self.arg_model.model_validate(arguments_pre_parsed)
62 | arguments_parsed_dict = arguments_parsed_model.model_dump_one_level()
63 |
64 | arguments_parsed_dict |= arguments_to_pass_directly or {}
65 |
66 | if fn_is_async:
67 | if isinstance(fn, Awaitable):
68 | return await fn
69 | return await fn(**arguments_parsed_dict)
70 | if isinstance(fn, Callable):
71 | return fn(**arguments_parsed_dict)
72 | raise TypeError("fn must be either Callable or Awaitable")
73 |
74 | def pre_parse_json(self, data: dict[str, Any]) -> dict[str, Any]:
75 | """Pre-parse data from JSON.
76 |
77 | Return a dict with same keys as input but with values parsed from JSON
78 | if appropriate.
79 |
80 | This is to handle cases like `["a", "b", "c"]` being passed in as JSON inside
81 | a string rather than an actual list. Claude desktop is prone to this - in fact
82 | it seems incapable of NOT doing this. For sub-models, it tends to pass
83 | dicts (JSON objects) as JSON strings, which can be pre-parsed here.
84 | """
85 | new_data = data.copy() # Shallow copy
86 | for field_name, field_info in self.arg_model.model_fields.items():
87 | if field_name not in data.keys():
88 | continue
89 | if isinstance(data[field_name], str):
90 | try:
91 | pre_parsed = json.loads(data[field_name])
92 | except json.JSONDecodeError:
93 | continue # Not JSON - skip
94 | if isinstance(pre_parsed, (str, int, float)):
95 | # This is likely that the raw value is e.g. `"hello"` which we
96 | # Should really be parsed as '"hello"' in Python - but if we parse
97 | # it as JSON it'll turn into just 'hello'. So we skip it.
98 | continue
99 | new_data[field_name] = pre_parsed
100 | assert new_data.keys() == data.keys()
101 | return new_data
102 |
103 | model_config = ConfigDict(
104 | arbitrary_types_allowed=True,
105 | )
106 |
107 |
108 | def func_metadata(func: Callable, skip_names: Sequence[str] = ()) -> FuncMetadata:
109 | """Given a function, return metadata including a pydantic model representing its signature.
110 |
111 | The use case for this is
112 | ```
113 | meta = func_to_pyd(func)
114 | validated_args = meta.arg_model.model_validate(some_raw_data_dict)
115 | return func(**validated_args.model_dump_one_level())
116 | ```
117 |
118 | **critically** it also provides pre-parse helper to attempt to parse things from JSON.
119 |
120 | Args:
121 | func: The function to convert to a pydantic model
122 | skip_names: A list of parameter names to skip. These will not be included in
123 | the model.
124 | Returns:
125 | A pydantic model representing the function's signature.
126 | """
127 | sig = _get_typed_signature(func)
128 | params = sig.parameters
129 | dynamic_pydantic_model_params: dict[str, Any] = {}
130 | globalns = getattr(func, "__globals__", {})
131 | for param in params.values():
132 | if param.name.startswith("_"):
133 | raise InvalidSignature(
134 | f"Parameter {param.name} of {func.__name__} may not start with an underscore"
135 | )
136 | if param.name in skip_names:
137 | continue
138 | annotation = param.annotation
139 |
140 | # `x: None` / `x: None = None`
141 | if annotation is None:
142 | annotation = Annotated[
143 | None,
144 | Field(
145 | default=param.default
146 | if param.default is not inspect.Parameter.empty
147 | else PydanticUndefined
148 | ),
149 | ]
150 |
151 | # Untyped field
152 | if annotation is inspect.Parameter.empty:
153 | annotation = Annotated[
154 | Any,
155 | Field(),
156 | # 🤷
157 | WithJsonSchema({"title": param.name, "type": "string"}),
158 | ]
159 |
160 | field_info = FieldInfo.from_annotated_attribute(
161 | _get_typed_annotation(annotation, globalns),
162 | param.default
163 | if param.default is not inspect.Parameter.empty
164 | else PydanticUndefined,
165 | )
166 | dynamic_pydantic_model_params[param.name] = (field_info.annotation, field_info)
167 | continue
168 |
169 | arguments_model = create_model(
170 | f"{func.__name__}Arguments",
171 | **dynamic_pydantic_model_params,
172 | __base__=ArgModelBase,
173 | )
174 | resp = FuncMetadata(arg_model=arguments_model)
175 | return resp
176 |
177 |
178 | def _get_typed_annotation(annotation: Any, globalns: Dict[str, Any]) -> Any:
179 | if isinstance(annotation, str):
180 | annotation = ForwardRef(annotation)
181 | annotation = eval_type_lenient(annotation, globalns, globalns)
182 |
183 | return annotation
184 |
185 |
186 | def _get_typed_signature(call: Callable[..., Any]) -> inspect.Signature:
187 | """Get function signature while evaluating forward references"""
188 | signature = inspect.signature(call)
189 | globalns = getattr(call, "__globals__", {})
190 | typed_params = [
191 | inspect.Parameter(
192 | name=param.name,
193 | kind=param.kind,
194 | default=param.default,
195 | annotation=_get_typed_annotation(param.annotation, globalns),
196 | )
197 | for param in signature.parameters.values()
198 | ]
199 | typed_signature = inspect.Signature(typed_params)
200 | return typed_signature
201 |
--------------------------------------------------------------------------------
/src/fastmcp/utilities/logging.py:
--------------------------------------------------------------------------------
1 | """Logging utilities for FastMCP."""
2 |
3 | import logging
4 | from typing import Literal
5 |
6 | from rich.console import Console
7 | from rich.logging import RichHandler
8 |
9 |
10 | def get_logger(name: str) -> logging.Logger:
11 | """Get a logger nested under FastMCP namespace.
12 |
13 | Args:
14 | name: the name of the logger, which will be prefixed with 'FastMCP.'
15 |
16 | Returns:
17 | a configured logger instance
18 | """
19 | return logging.getLogger(f"FastMCP.{name}")
20 |
21 |
22 | def configure_logging(
23 | level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO",
24 | ) -> None:
25 | """Configure logging for FastMCP.
26 |
27 | Args:
28 | level: the log level to use
29 | """
30 | logging.basicConfig(
31 | level=level,
32 | format="%(message)s",
33 | handlers=[RichHandler(console=Console(stderr=True), rich_tracebacks=True)],
34 | )
35 |
--------------------------------------------------------------------------------
/src/fastmcp/utilities/types.py:
--------------------------------------------------------------------------------
1 | """Common types used across FastMCP."""
2 |
3 | import base64
4 | from pathlib import Path
5 | from typing import Optional, Union
6 |
7 | from mcp.types import ImageContent
8 |
9 |
10 | class Image:
11 | """Helper class for returning images from tools."""
12 |
13 | def __init__(
14 | self,
15 | path: Optional[Union[str, Path]] = None,
16 | data: Optional[bytes] = None,
17 | format: Optional[str] = None,
18 | ):
19 | if path is None and data is None:
20 | raise ValueError("Either path or data must be provided")
21 | if path is not None and data is not None:
22 | raise ValueError("Only one of path or data can be provided")
23 |
24 | self.path = Path(path) if path else None
25 | self.data = data
26 | self._format = format
27 | self._mime_type = self._get_mime_type()
28 |
29 | def _get_mime_type(self) -> str:
30 | """Get MIME type from format or guess from file extension."""
31 | if self._format:
32 | return f"image/{self._format.lower()}"
33 |
34 | if self.path:
35 | suffix = self.path.suffix.lower()
36 | return {
37 | ".png": "image/png",
38 | ".jpg": "image/jpeg",
39 | ".jpeg": "image/jpeg",
40 | ".gif": "image/gif",
41 | ".webp": "image/webp",
42 | }.get(suffix, "application/octet-stream")
43 | return "image/png" # default for raw binary data
44 |
45 | def to_image_content(self) -> ImageContent:
46 | """Convert to MCP ImageContent."""
47 | if self.path:
48 | with open(self.path, "rb") as f:
49 | data = base64.b64encode(f.read()).decode()
50 | elif self.data is not None:
51 | data = base64.b64encode(self.data).decode()
52 | else:
53 | raise ValueError("No image data available")
54 |
55 | return ImageContent(type="image", data=data, mimeType=self._mime_type)
56 |
--------------------------------------------------------------------------------
/tests/__init__.py:
--------------------------------------------------------------------------------
https://raw.githubusercontent.com/jlowin/fastmcp/80c328b3dc0b949f010456ee0e85cc5c90e3305f/tests/__init__.py
--------------------------------------------------------------------------------
/tests/prompts/__init__.py:
--------------------------------------------------------------------------------
https://raw.githubusercontent.com/jlowin/fastmcp/80c328b3dc0b949f010456ee0e85cc5c90e3305f/tests/prompts/__init__.py
--------------------------------------------------------------------------------
/tests/prompts/test_base.py:
--------------------------------------------------------------------------------
1 | from pydantic import FileUrl
2 | import pytest
3 | from fastmcp.prompts.base import (
4 | Prompt,
5 | UserMessage,
6 | TextContent,
7 | AssistantMessage,
8 | Message,
9 | )
10 | from mcp.types import EmbeddedResource, TextResourceContents
11 |
12 |
13 | class TestRenderPrompt:
14 | async def test_basic_fn(self):
15 | def fn() -> str:
16 | return "Hello, world!"
17 |
18 | prompt = Prompt.from_function(fn)
19 | assert await prompt.render() == [
20 | UserMessage(content=TextContent(type="text", text="Hello, world!"))
21 | ]
22 |
23 | async def test_async_fn(self):
24 | async def fn() -> str:
25 | return "Hello, world!"
26 |
27 | prompt = Prompt.from_function(fn)
28 | assert await prompt.render() == [
29 | UserMessage(content=TextContent(type="text", text="Hello, world!"))
30 | ]
31 |
32 | async def test_fn_with_args(self):
33 | async def fn(name: str, age: int = 30) -> str:
34 | return f"Hello, {name}! You're {age} years old."
35 |
36 | prompt = Prompt.from_function(fn)
37 | assert await prompt.render(arguments=dict(name="World")) == [
38 | UserMessage(
39 | content=TextContent(
40 | type="text", text="Hello, World! You're 30 years old."
41 | )
42 | )
43 | ]
44 |
45 | async def test_fn_with_invalid_kwargs(self):
46 | async def fn(name: str, age: int = 30) -> str:
47 | return f"Hello, {name}! You're {age} years old."
48 |
49 | prompt = Prompt.from_function(fn)
50 | with pytest.raises(ValueError):
51 | await prompt.render(arguments=dict(age=40))
52 |
53 | async def test_fn_returns_message(self):
54 | async def fn() -> UserMessage:
55 | return UserMessage(content="Hello, world!")
56 |
57 | prompt = Prompt.from_function(fn)
58 | assert await prompt.render() == [
59 | UserMessage(content=TextContent(type="text", text="Hello, world!"))
60 | ]
61 |
62 | async def test_fn_returns_assistant_message(self):
63 | async def fn() -> AssistantMessage:
64 | return AssistantMessage(
65 | content=TextContent(type="text", text="Hello, world!")
66 | )
67 |
68 | prompt = Prompt.from_function(fn)
69 | assert await prompt.render() == [
70 | AssistantMessage(content=TextContent(type="text", text="Hello, world!"))
71 | ]
72 |
73 | async def test_fn_returns_multiple_messages(self):
74 | expected = [
75 | UserMessage("Hello, world!"),
76 | AssistantMessage("How can I help you today?"),
77 | UserMessage("I'm looking for a restaurant in the center of town."),
78 | ]
79 |
80 | async def fn() -> list[Message]:
81 | return expected
82 |
83 | prompt = Prompt.from_function(fn)
84 | assert await prompt.render() == expected
85 |
86 | async def test_fn_returns_list_of_strings(self):
87 | expected = [
88 | "Hello, world!",
89 | "I'm looking for a restaurant in the center of town.",
90 | ]
91 |
92 | async def fn() -> list[str]:
93 | return expected
94 |
95 | prompt = Prompt.from_function(fn)
96 | assert await prompt.render() == [UserMessage(t) for t in expected]
97 |
98 | async def test_fn_returns_resource_content(self):
99 | """Test returning a message with resource content."""
100 |
101 | async def fn() -> UserMessage:
102 | return UserMessage(
103 | content=EmbeddedResource(
104 | type="resource",
105 | resource=TextResourceContents(
106 | uri=FileUrl("file://file.txt"),
107 | text="File contents",
108 | mimeType="text/plain",
109 | ),
110 | )
111 | )
112 |
113 | prompt = Prompt.from_function(fn)
114 | assert await prompt.render() == [
115 | UserMessage(
116 | content=EmbeddedResource(
117 | type="resource",
118 | resource=TextResourceContents(
119 | uri=FileUrl("file://file.txt"),
120 | text="File contents",
121 | mimeType="text/plain",
122 | ),
123 | )
124 | )
125 | ]
126 |
127 | async def test_fn_returns_mixed_content(self):
128 | """Test returning messages with mixed content types."""
129 |
130 | async def fn() -> list[Message]:
131 | return [
132 | UserMessage(content="Please analyze this file:"),
133 | UserMessage(
134 | content=EmbeddedResource(
135 | type="resource",
136 | resource=TextResourceContents(
137 | uri=FileUrl("file://file.txt"),
138 | text="File contents",
139 | mimeType="text/plain",
140 | ),
141 | )
142 | ),
143 | AssistantMessage(content="I'll help analyze that file."),
144 | ]
145 |
146 | prompt = Prompt.from_function(fn)
147 | assert await prompt.render() == [
148 | UserMessage(
149 | content=TextContent(type="text", text="Please analyze this file:")
150 | ),
151 | UserMessage(
152 | content=EmbeddedResource(
153 | type="resource",
154 | resource=TextResourceContents(
155 | uri=FileUrl("file://file.txt"),
156 | text="File contents",
157 | mimeType="text/plain",
158 | ),
159 | )
160 | ),
161 | AssistantMessage(
162 | content=TextContent(type="text", text="I'll help analyze that file.")
163 | ),
164 | ]
165 |
166 | async def test_fn_returns_dict_with_resource(self):
167 | """Test returning a dict with resource content."""
168 |
169 | async def fn() -> dict:
170 | return {
171 | "role": "user",
172 | "content": {
173 | "type": "resource",
174 | "resource": {
175 | "uri": FileUrl("file://file.txt"),
176 | "text": "File contents",
177 | "mimeType": "text/plain",
178 | },
179 | },
180 | }
181 |
182 | prompt = Prompt.from_function(fn)
183 | assert await prompt.render() == [
184 | UserMessage(
185 | content=EmbeddedResource(
186 | type="resource",
187 | resource=TextResourceContents(
188 | uri=FileUrl("file://file.txt"),
189 | text="File contents",
190 | mimeType="text/plain",
191 | ),
192 | )
193 | )
194 | ]
195 |
--------------------------------------------------------------------------------
/tests/prompts/test_manager.py:
--------------------------------------------------------------------------------
1 | import pytest
2 | from fastmcp.prompts.base import UserMessage, TextContent, Prompt
3 | from fastmcp.prompts.manager import PromptManager
4 |
5 |
6 | class TestPromptManager:
7 | def test_add_prompt(self):
8 | """Test adding a prompt to the manager."""
9 |
10 | def fn() -> str:
11 | return "Hello, world!"
12 |
13 | manager = PromptManager()
14 | prompt = Prompt.from_function(fn)
15 | added = manager.add_prompt(prompt)
16 | assert added == prompt
17 | assert manager.get_prompt("fn") == prompt
18 |
19 | def test_add_duplicate_prompt(self, caplog):
20 | """Test adding the same prompt twice."""
21 |
22 | def fn() -> str:
23 | return "Hello, world!"
24 |
25 | manager = PromptManager()
26 | prompt = Prompt.from_function(fn)
27 | first = manager.add_prompt(prompt)
28 | second = manager.add_prompt(prompt)
29 | assert first == second
30 | assert "Prompt already exists" in caplog.text
31 |
32 | def test_disable_warn_on_duplicate_prompts(self, caplog):
33 | """Test disabling warning on duplicate prompts."""
34 |
35 | def fn() -> str:
36 | return "Hello, world!"
37 |
38 | manager = PromptManager(warn_on_duplicate_prompts=False)
39 | prompt = Prompt.from_function(fn)
40 | first = manager.add_prompt(prompt)
41 | second = manager.add_prompt(prompt)
42 | assert first == second
43 | assert "Prompt already exists" not in caplog.text
44 |
45 | def test_list_prompts(self):
46 | """Test listing all prompts."""
47 |
48 | def fn1() -> str:
49 | return "Hello, world!"
50 |
51 | def fn2() -> str:
52 | return "Goodbye, world!"
53 |
54 | manager = PromptManager()
55 | prompt1 = Prompt.from_function(fn1)
56 | prompt2 = Prompt.from_function(fn2)
57 | manager.add_prompt(prompt1)
58 | manager.add_prompt(prompt2)
59 | prompts = manager.list_prompts()
60 | assert len(prompts) == 2
61 | assert prompts == [prompt1, prompt2]
62 |
63 | async def test_render_prompt(self):
64 | """Test rendering a prompt."""
65 |
66 | def fn() -> str:
67 | return "Hello, world!"
68 |
69 | manager = PromptManager()
70 | prompt = Prompt.from_function(fn)
71 | manager.add_prompt(prompt)
72 | messages = await manager.render_prompt("fn")
73 | assert messages == [
74 | UserMessage(content=TextContent(type="text", text="Hello, world!"))
75 | ]
76 |
77 | async def test_render_prompt_with_args(self):
78 | """Test rendering a prompt with arguments."""
79 |
80 | def fn(name: str) -> str:
81 | return f"Hello, {name}!"
82 |
83 | manager = PromptManager()
84 | prompt = Prompt.from_function(fn)
85 | manager.add_prompt(prompt)
86 | messages = await manager.render_prompt("fn", arguments={"name": "World"})
87 | assert messages == [
88 | UserMessage(content=TextContent(type="text", text="Hello, World!"))
89 | ]
90 |
91 | async def test_render_unknown_prompt(self):
92 | """Test rendering a non-existent prompt."""
93 | manager = PromptManager()
94 | with pytest.raises(ValueError, match="Unknown prompt: unknown"):
95 | await manager.render_prompt("unknown")
96 |
97 | async def test_render_prompt_with_missing_args(self):
98 | """Test rendering a prompt with missing required arguments."""
99 |
100 | def fn(name: str) -> str:
101 | return f"Hello, {name}!"
102 |
103 | manager = PromptManager()
104 | prompt = Prompt.from_function(fn)
105 | manager.add_prompt(prompt)
106 | with pytest.raises(ValueError, match="Missing required arguments"):
107 | await manager.render_prompt("fn")
108 |
--------------------------------------------------------------------------------
/tests/resources/__init__.py:
--------------------------------------------------------------------------------
https://raw.githubusercontent.com/jlowin/fastmcp/80c328b3dc0b949f010456ee0e85cc5c90e3305f/tests/resources/__init__.py
--------------------------------------------------------------------------------
/tests/resources/test_file_resources.py:
--------------------------------------------------------------------------------
1 | import os
2 |
3 | import pytest
4 | from pathlib import Path
5 | from tempfile import NamedTemporaryFile
6 | from pydantic import FileUrl
7 |
8 | from fastmcp.resources import FileResource
9 |
10 |
11 | @pytest.fixture
12 | def temp_file():
13 | """Create a temporary file for testing.
14 |
15 | File is automatically cleaned up after the test if it still exists.
16 | """
17 | content = "test content"
18 | with NamedTemporaryFile(mode="w", delete=False) as f:
19 | f.write(content)
20 | path = Path(f.name).resolve()
21 | yield path
22 | try:
23 | path.unlink()
24 | except FileNotFoundError:
25 | pass # File was already deleted by the test
26 |
27 |
28 | class TestFileResource:
29 | """Test FileResource functionality."""
30 |
31 | def test_file_resource_creation(self, temp_file: Path):
32 | """Test creating a FileResource."""
33 | resource = FileResource(
34 | uri=FileUrl(temp_file.as_uri()),
35 | name="test",
36 | description="test file",
37 | path=temp_file,
38 | )
39 | assert str(resource.uri) == temp_file.as_uri()
40 | assert resource.name == "test"
41 | assert resource.description == "test file"
42 | assert resource.mime_type == "text/plain" # default
43 | assert resource.path == temp_file
44 | assert resource.is_binary is False # default
45 |
46 | def test_file_resource_str_path_conversion(self, temp_file: Path):
47 | """Test FileResource handles string paths."""
48 | resource = FileResource(
49 | uri=FileUrl(f"file://{temp_file}"),
50 | name="test",
51 | path=Path(str(temp_file)),
52 | )
53 | assert isinstance(resource.path, Path)
54 | assert resource.path.is_absolute()
55 |
56 | async def test_read_text_file(self, temp_file: Path):
57 | """Test reading a text file."""
58 | resource = FileResource(
59 | uri=FileUrl(f"file://{temp_file}"),
60 | name="test",
61 | path=temp_file,
62 | )
63 | content = await resource.read()
64 | assert content == "test content"
65 | assert resource.mime_type == "text/plain"
66 |
67 | async def test_read_binary_file(self, temp_file: Path):
68 | """Test reading a file as binary."""
69 | resource = FileResource(
70 | uri=FileUrl(f"file://{temp_file}"),
71 | name="test",
72 | path=temp_file,
73 | is_binary=True,
74 | )
75 | content = await resource.read()
76 | assert isinstance(content, bytes)
77 | assert content == b"test content"
78 |
79 | def test_relative_path_error(self):
80 | """Test error on relative path."""
81 | with pytest.raises(ValueError, match="Path must be absolute"):
82 | FileResource(
83 | uri=FileUrl("file:///test.txt"),
84 | name="test",
85 | path=Path("test.txt"),
86 | )
87 |
88 | async def test_missing_file_error(self, temp_file: Path):
89 | """Test error when file doesn't exist."""
90 | # Create path to non-existent file
91 | missing = temp_file.parent / "missing.txt"
92 | resource = FileResource(
93 | uri=FileUrl("file:///missing.txt"),
94 | name="test",
95 | path=missing,
96 | )
97 | with pytest.raises(ValueError, match="Error reading file"):
98 | await resource.read()
99 |
100 | @pytest.mark.skipif(
101 | os.name == "nt", reason="File permissions behave differently on Windows"
102 | )
103 | async def test_permission_error(self, temp_file: Path):
104 | """Test reading a file without permissions."""
105 | temp_file.chmod(0o000) # Remove all permissions
106 | try:
107 | resource = FileResource(
108 | uri=FileUrl(temp_file.as_uri()),
109 | name="test",
110 | path=temp_file,
111 | )
112 | with pytest.raises(ValueError, match="Error reading file"):
113 | await resource.read()
114 | finally:
115 | temp_file.chmod(0o644) # Restore permissions
116 |
--------------------------------------------------------------------------------
/tests/resources/test_function_resources.py:
--------------------------------------------------------------------------------
1 | from pydantic import BaseModel, AnyUrl
2 | import pytest
3 | from fastmcp.resources import FunctionResource
4 |
5 |
6 | class TestFunctionResource:
7 | """Test FunctionResource functionality."""
8 |
9 | def test_function_resource_creation(self):
10 | """Test creating a FunctionResource."""
11 |
12 | def my_func() -> str:
13 | return "test content"
14 |
15 | resource = FunctionResource(
16 | uri=AnyUrl("fn://test"),
17 | name="test",
18 | description="test function",
19 | fn=my_func,
20 | )
21 | assert str(resource.uri) == "fn://test"
22 | assert resource.name == "test"
23 | assert resource.description == "test function"
24 | assert resource.mime_type == "text/plain" # default
25 | assert resource.fn == my_func
26 |
27 | async def test_read_text(self):
28 | """Test reading text from a FunctionResource."""
29 |
30 | def get_data() -> str:
31 | return "Hello, world!"
32 |
33 | resource = FunctionResource(
34 | uri=AnyUrl("function://test"),
35 | name="test",
36 | fn=get_data,
37 | )
38 | content = await resource.read()
39 | assert content == "Hello, world!"
40 | assert resource.mime_type == "text/plain"
41 |
42 | async def test_read_binary(self):
43 | """Test reading binary data from a FunctionResource."""
44 |
45 | def get_data() -> bytes:
46 | return b"Hello, world!"
47 |
48 | resource = FunctionResource(
49 | uri=AnyUrl("function://test"),
50 | name="test",
51 | fn=get_data,
52 | )
53 | content = await resource.read()
54 | assert content == b"Hello, world!"
55 |
56 | async def test_json_conversion(self):
57 | """Test automatic JSON conversion of non-string results."""
58 |
59 | def get_data() -> dict:
60 | return {"key": "value"}
61 |
62 | resource = FunctionResource(
63 | uri=AnyUrl("function://test"),
64 | name="test",
65 | fn=get_data,
66 | )
67 | content = await resource.read()
68 | assert isinstance(content, str)
69 | assert '"key": "value"' in content
70 |
71 | async def test_error_handling(self):
72 | """Test error handling in FunctionResource."""
73 |
74 | def failing_func() -> str:
75 | raise ValueError("Test error")
76 |
77 | resource = FunctionResource(
78 | uri=AnyUrl("function://test"),
79 | name="test",
80 | fn=failing_func,
81 | )
82 | with pytest.raises(ValueError, match="Error reading resource function://test"):
83 | await resource.read()
84 |
85 | async def test_basemodel_conversion(self):
86 | """Test handling of BaseModel types."""
87 |
88 | class MyModel(BaseModel):
89 | name: str
90 |
91 | resource = FunctionResource(
92 | uri=AnyUrl("function://test"),
93 | name="test",
94 | fn=lambda: MyModel(name="test"),
95 | )
96 | content = await resource.read()
97 | assert content == '{"name": "test"}'
98 |
99 | async def test_custom_type_conversion(self):
100 | """Test handling of custom types."""
101 |
102 | class CustomData:
103 | def __str__(self) -> str:
104 | return "custom data"
105 |
106 | def get_data() -> CustomData:
107 | return CustomData()
108 |
109 | resource = FunctionResource(
110 | uri=AnyUrl("function://test"),
111 | name="test",
112 | fn=get_data,
113 | )
114 | content = await resource.read()
115 | assert isinstance(content, str)
116 |
--------------------------------------------------------------------------------
/tests/resources/test_resource_manager.py:
--------------------------------------------------------------------------------
1 | import pytest
2 | from pathlib import Path
3 | from tempfile import NamedTemporaryFile
4 | from pydantic import AnyUrl, FileUrl
5 |
6 | from fastmcp.resources import (
7 | FileResource,
8 | FunctionResource,
9 | ResourceManager,
10 | ResourceTemplate,
11 | )
12 |
13 |
14 | @pytest.fixture
15 | def temp_file():
16 | """Create a temporary file for testing.
17 |
18 | File is automatically cleaned up after the test if it still exists.
19 | """
20 | content = "test content"
21 | with NamedTemporaryFile(mode="w", delete=False) as f:
22 | f.write(content)
23 | path = Path(f.name).resolve()
24 | yield path
25 | try:
26 | path.unlink()
27 | except FileNotFoundError:
28 | pass # File was already deleted by the test
29 |
30 |
31 | class TestResourceManager:
32 | """Test ResourceManager functionality."""
33 |
34 | def test_add_resource(self, temp_file: Path):
35 | """Test adding a resource."""
36 | manager = ResourceManager()
37 | resource = FileResource(
38 | uri=FileUrl(f"file://{temp_file}"),
39 | name="test",
40 | path=temp_file,
41 | )
42 | added = manager.add_resource(resource)
43 | assert added == resource
44 | assert manager.list_resources() == [resource]
45 |
46 | def test_add_duplicate_resource(self, temp_file: Path):
47 | """Test adding the same resource twice."""
48 | manager = ResourceManager()
49 | resource = FileResource(
50 | uri=FileUrl(f"file://{temp_file}"),
51 | name="test",
52 | path=temp_file,
53 | )
54 | first = manager.add_resource(resource)
55 | second = manager.add_resource(resource)
56 | assert first == second
57 | assert manager.list_resources() == [resource]
58 |
59 | def test_warn_on_duplicate_resources(self, temp_file: Path, caplog):
60 | """Test warning on duplicate resources."""
61 | manager = ResourceManager()
62 | resource = FileResource(
63 | uri=FileUrl(f"file://{temp_file}"),
64 | name="test",
65 | path=temp_file,
66 | )
67 | manager.add_resource(resource)
68 | manager.add_resource(resource)
69 | assert "Resource already exists" in caplog.text
70 |
71 | def test_disable_warn_on_duplicate_resources(self, temp_file: Path, caplog):
72 | """Test disabling warning on duplicate resources."""
73 | manager = ResourceManager(warn_on_duplicate_resources=False)
74 | resource = FileResource(
75 | uri=FileUrl(f"file://{temp_file}"),
76 | name="test",
77 | path=temp_file,
78 | )
79 | manager.add_resource(resource)
80 | manager.add_resource(resource)
81 | assert "Resource already exists" not in caplog.text
82 |
83 | async def test_get_resource(self, temp_file: Path):
84 | """Test getting a resource by URI."""
85 | manager = ResourceManager()
86 | resource = FileResource(
87 | uri=FileUrl(f"file://{temp_file}"),
88 | name="test",
89 | path=temp_file,
90 | )
91 | manager.add_resource(resource)
92 | retrieved = await manager.get_resource(resource.uri)
93 | assert retrieved == resource
94 |
95 | async def test_get_resource_from_template(self):
96 | """Test getting a resource through a template."""
97 | manager = ResourceManager()
98 |
99 | def greet(name: str) -> str:
100 | return f"Hello, {name}!"
101 |
102 | template = ResourceTemplate.from_function(
103 | fn=greet,
104 | uri_template="greet://{name}",
105 | name="greeter",
106 | )
107 | manager._templates[template.uri_template] = template
108 |
109 | resource = await manager.get_resource(AnyUrl("greet://world"))
110 | assert isinstance(resource, FunctionResource)
111 | content = await resource.read()
112 | assert content == "Hello, world!"
113 |
114 | async def test_get_unknown_resource(self):
115 | """Test getting a non-existent resource."""
116 | manager = ResourceManager()
117 | with pytest.raises(ValueError, match="Unknown resource"):
118 | await manager.get_resource(AnyUrl("unknown://test"))
119 |
120 | def test_list_resources(self, temp_file: Path):
121 | """Test listing all resources."""
122 | manager = ResourceManager()
123 | resource1 = FileResource(
124 | uri=FileUrl(f"file://{temp_file}"),
125 | name="test1",
126 | path=temp_file,
127 | )
128 | resource2 = FileResource(
129 | uri=FileUrl(f"file://{temp_file}2"),
130 | name="test2",
131 | path=temp_file,
132 | )
133 | manager.add_resource(resource1)
134 | manager.add_resource(resource2)
135 | resources = manager.list_resources()
136 | assert len(resources) == 2
137 | assert resources == [resource1, resource2]
138 |
--------------------------------------------------------------------------------
/tests/resources/test_resource_template.py:
--------------------------------------------------------------------------------
1 | import json
2 | import pytest
3 | from pydantic import BaseModel
4 |
5 | from fastmcp.resources import FunctionResource, ResourceTemplate
6 |
7 |
8 | class TestResourceTemplate:
9 | """Test ResourceTemplate functionality."""
10 |
11 | def test_template_creation(self):
12 | """Test creating a template from a function."""
13 |
14 | def my_func(key: str, value: int) -> dict:
15 | return {"key": key, "value": value}
16 |
17 | template = ResourceTemplate.from_function(
18 | fn=my_func,
19 | uri_template="test://{key}/{value}",
20 | name="test",
21 | )
22 | assert template.uri_template == "test://{key}/{value}"
23 | assert template.name == "test"
24 | assert template.mime_type == "text/plain" # default
25 | test_input = {"key": "test", "value": 42}
26 | assert template.fn(**test_input) == my_func(**test_input)
27 |
28 | def test_template_matches(self):
29 | """Test matching URIs against a template."""
30 |
31 | def my_func(key: str, value: int) -> dict:
32 | return {"key": key, "value": value}
33 |
34 | template = ResourceTemplate.from_function(
35 | fn=my_func,
36 | uri_template="test://{key}/{value}",
37 | name="test",
38 | )
39 |
40 | # Valid match
41 | params = template.matches("test://foo/123")
42 | assert params == {"key": "foo", "value": "123"}
43 |
44 | # No match
45 | assert template.matches("test://foo") is None
46 | assert template.matches("other://foo/123") is None
47 |
48 | async def test_create_resource(self):
49 | """Test creating a resource from a template."""
50 |
51 | def my_func(key: str, value: int) -> dict:
52 | return {"key": key, "value": value}
53 |
54 | template = ResourceTemplate.from_function(
55 | fn=my_func,
56 | uri_template="test://{key}/{value}",
57 | name="test",
58 | )
59 |
60 | resource = await template.create_resource(
61 | "test://foo/123",
62 | {"key": "foo", "value": 123},
63 | )
64 |
65 | assert isinstance(resource, FunctionResource)
66 | content = await resource.read()
67 | assert isinstance(content, str)
68 | data = json.loads(content)
69 | assert data == {"key": "foo", "value": 123}
70 |
71 | async def test_template_error(self):
72 | """Test error handling in template resource creation."""
73 |
74 | def failing_func(x: str) -> str:
75 | raise ValueError("Test error")
76 |
77 | template = ResourceTemplate.from_function(
78 | fn=failing_func,
79 | uri_template="fail://{x}",
80 | name="fail",
81 | )
82 |
83 | with pytest.raises(ValueError, match="Error creating resource from template"):
84 | await template.create_resource("fail://test", {"x": "test"})
85 |
86 | async def test_async_text_resource(self):
87 | """Test creating a text resource from async function."""
88 |
89 | async def greet(name: str) -> str:
90 | return f"Hello, {name}!"
91 |
92 | template = ResourceTemplate.from_function(
93 | fn=greet,
94 | uri_template="greet://{name}",
95 | name="greeter",
96 | )
97 |
98 | resource = await template.create_resource(
99 | "greet://world",
100 | {"name": "world"},
101 | )
102 |
103 | assert isinstance(resource, FunctionResource)
104 | content = await resource.read()
105 | assert content == "Hello, world!"
106 |
107 | async def test_async_binary_resource(self):
108 | """Test creating a binary resource from async function."""
109 |
110 | async def get_bytes(value: str) -> bytes:
111 | return value.encode()
112 |
113 | template = ResourceTemplate.from_function(
114 | fn=get_bytes,
115 | uri_template="bytes://{value}",
116 | name="bytes",
117 | )
118 |
119 | resource = await template.create_resource(
120 | "bytes://test",
121 | {"value": "test"},
122 | )
123 |
124 | assert isinstance(resource, FunctionResource)
125 | content = await resource.read()
126 | assert content == b"test"
127 |
128 | async def test_basemodel_conversion(self):
129 | """Test handling of BaseModel types."""
130 |
131 | class MyModel(BaseModel):
132 | key: str
133 | value: int
134 |
135 | def get_data(key: str, value: int) -> MyModel:
136 | return MyModel(key=key, value=value)
137 |
138 | template = ResourceTemplate.from_function(
139 | fn=get_data,
140 | uri_template="test://{key}/{value}",
141 | name="test",
142 | )
143 |
144 | resource = await template.create_resource(
145 | "test://foo/123",
146 | {"key": "foo", "value": 123},
147 | )
148 |
149 | assert isinstance(resource, FunctionResource)
150 | content = await resource.read()
151 | assert isinstance(content, str)
152 | data = json.loads(content)
153 | assert data == {"key": "foo", "value": 123}
154 |
155 | async def test_custom_type_conversion(self):
156 | """Test handling of custom types."""
157 |
158 | class CustomData:
159 | def __init__(self, value: str):
160 | self.value = value
161 |
162 | def __str__(self) -> str:
163 | return self.value
164 |
165 | def get_data(value: str) -> CustomData:
166 | return CustomData(value)
167 |
168 | template = ResourceTemplate.from_function(
169 | fn=get_data,
170 | uri_template="test://{value}",
171 | name="test",
172 | )
173 |
174 | resource = await template.create_resource(
175 | "test://hello",
176 | {"value": "hello"},
177 | )
178 |
179 | assert isinstance(resource, FunctionResource)
180 | content = await resource.read()
181 | assert content == "hello"
182 |
--------------------------------------------------------------------------------
/tests/resources/test_resources.py:
--------------------------------------------------------------------------------
1 | import pytest
2 | from pydantic import AnyUrl
3 |
4 | from fastmcp.resources import FunctionResource, Resource
5 |
6 |
7 | class TestResourceValidation:
8 | """Test base Resource validation."""
9 |
10 | def test_resource_uri_validation(self):
11 | """Test URI validation."""
12 |
13 | def dummy_func() -> str:
14 | return "data"
15 |
16 | # Valid URI
17 | resource = FunctionResource(
18 | uri=AnyUrl("http://example.com/data"),
19 | name="test",
20 | fn=dummy_func,
21 | )
22 | assert str(resource.uri) == "http://example.com/data"
23 |
24 | # Missing protocol
25 | with pytest.raises(ValueError, match="Input should be a valid URL"):
26 | FunctionResource(
27 | uri=AnyUrl("invalid"),
28 | name="test",
29 | fn=dummy_func,
30 | )
31 |
32 | # Missing host
33 | with pytest.raises(ValueError, match="Input should be a valid URL"):
34 | FunctionResource(
35 | uri=AnyUrl("http://"),
36 | name="test",
37 | fn=dummy_func,
38 | )
39 |
40 | def test_resource_name_from_uri(self):
41 | """Test name is extracted from URI if not provided."""
42 |
43 | def dummy_func() -> str:
44 | return "data"
45 |
46 | resource = FunctionResource(
47 | uri=AnyUrl("resource://my-resource"),
48 | fn=dummy_func,
49 | )
50 | assert resource.name == "resource://my-resource"
51 |
52 | def test_resource_name_validation(self):
53 | """Test name validation."""
54 |
55 | def dummy_func() -> str:
56 | return "data"
57 |
58 | # Must provide either name or URI
59 | with pytest.raises(ValueError, match="Either name or uri must be provided"):
60 | FunctionResource(
61 | fn=dummy_func,
62 | )
63 |
64 | # Explicit name takes precedence over URI
65 | resource = FunctionResource(
66 | uri=AnyUrl("resource://uri-name"),
67 | name="explicit-name",
68 | fn=dummy_func,
69 | )
70 | assert resource.name == "explicit-name"
71 |
72 | def test_resource_mime_type(self):
73 | """Test mime type handling."""
74 |
75 | def dummy_func() -> str:
76 | return "data"
77 |
78 | # Default mime type
79 | resource = FunctionResource(
80 | uri=AnyUrl("resource://test"),
81 | fn=dummy_func,
82 | )
83 | assert resource.mime_type == "text/plain"
84 |
85 | # Custom mime type
86 | resource = FunctionResource(
87 | uri=AnyUrl("resource://test"),
88 | fn=dummy_func,
89 | mime_type="application/json",
90 | )
91 | assert resource.mime_type == "application/json"
92 |
93 | async def test_resource_read_abstract(self):
94 | """Test that Resource.read() is abstract."""
95 |
96 | class ConcreteResource(Resource):
97 | pass
98 |
99 | with pytest.raises(TypeError, match="abstract method"):
100 | ConcreteResource(uri=AnyUrl("test://test"), name="test") # type: ignore
101 |
--------------------------------------------------------------------------------
/tests/servers/__init__.py:
--------------------------------------------------------------------------------
https://raw.githubusercontent.com/jlowin/fastmcp/80c328b3dc0b949f010456ee0e85cc5c90e3305f/tests/servers/__init__.py
--------------------------------------------------------------------------------
/tests/servers/test_file_server.py:
--------------------------------------------------------------------------------
1 | import json
2 | from fastmcp import FastMCP
3 | import pytest
4 | from pathlib import Path
5 |
6 |
7 | @pytest.fixture()
8 | def test_dir(tmp_path_factory) -> Path:
9 | """Create a temporary directory with test files."""
10 | tmp = tmp_path_factory.mktemp("test_files")
11 |
12 | # Create test files
13 | (tmp / "example.py").write_text("print('hello world')")
14 | (tmp / "readme.md").write_text("# Test Directory\nThis is a test.")
15 | (tmp / "config.json").write_text('{"test": true}')
16 |
17 | return tmp
18 |
19 |
20 | @pytest.fixture
21 | def mcp() -> FastMCP:
22 | mcp = FastMCP()
23 |
24 | return mcp
25 |
26 |
27 | @pytest.fixture(autouse=True)
28 | def resources(mcp: FastMCP, test_dir: Path) -> FastMCP:
29 | @mcp.resource("dir://test_dir")
30 | def list_test_dir() -> list[str]:
31 | """List the files in the test directory"""
32 | return [str(f) for f in test_dir.iterdir()]
33 |
34 | @mcp.resource("file://test_dir/example.py")
35 | def read_example_py() -> str:
36 | """Read the example.py file"""
37 | try:
38 | return (test_dir / "example.py").read_text()
39 | except FileNotFoundError:
40 | return "File not found"
41 |
42 | @mcp.resource("file://test_dir/readme.md")
43 | def read_readme_md() -> str:
44 | """Read the readme.md file"""
45 | try:
46 | return (test_dir / "readme.md").read_text()
47 | except FileNotFoundError:
48 | return "File not found"
49 |
50 | @mcp.resource("file://test_dir/config.json")
51 | def read_config_json() -> str:
52 | """Read the config.json file"""
53 | try:
54 | return (test_dir / "config.json").read_text()
55 | except FileNotFoundError:
56 | return "File not found"
57 |
58 | return mcp
59 |
60 |
61 | @pytest.fixture(autouse=True)
62 | def tools(mcp: FastMCP, test_dir: Path) -> FastMCP:
63 | @mcp.tool()
64 | def delete_file(path: str) -> bool:
65 | # ensure path is in test_dir
66 | if Path(path).resolve().parent != test_dir:
67 | raise ValueError(f"Path must be in test_dir: {path}")
68 | Path(path).unlink()
69 | return True
70 |
71 | return mcp
72 |
73 |
74 | async def test_list_resources(mcp: FastMCP):
75 | resources = await mcp.list_resources()
76 | assert len(resources) == 4
77 |
78 | assert [str(r.uri) for r in resources] == [
79 | "dir://test_dir",
80 | "file://test_dir/example.py",
81 | "file://test_dir/readme.md",
82 | "file://test_dir/config.json",
83 | ]
84 |
85 |
86 | async def test_read_resource_dir(mcp: FastMCP):
87 | files = await mcp.read_resource("dir://test_dir")
88 | files = json.loads(files)
89 |
90 | assert sorted([Path(f).name for f in files]) == [
91 | "config.json",
92 | "example.py",
93 | "readme.md",
94 | ]
95 |
96 |
97 | async def test_read_resource_file(mcp: FastMCP):
98 | result = await mcp.read_resource("file://test_dir/example.py")
99 | assert result == "print('hello world')"
100 |
101 |
102 | async def test_delete_file(mcp: FastMCP, test_dir: Path):
103 | await mcp.call_tool(
104 | "delete_file", arguments=dict(path=str(test_dir / "example.py"))
105 | )
106 | assert not (test_dir / "example.py").exists()
107 |
108 |
109 | async def test_delete_file_and_check_resources(mcp: FastMCP, test_dir: Path):
110 | await mcp.call_tool(
111 | "delete_file", arguments=dict(path=str(test_dir / "example.py"))
112 | )
113 | result = await mcp.read_resource("file://test_dir/example.py")
114 | assert result == "File not found"
115 |
--------------------------------------------------------------------------------
/tests/test_cli.py:
--------------------------------------------------------------------------------
1 | """Tests for the FastMCP CLI."""
2 |
3 | import json
4 | import sys
5 | from pathlib import Path
6 | from unittest.mock import call, patch
7 |
8 | import pytest
9 | from typer.testing import CliRunner
10 |
11 | from fastmcp.cli.cli import _parse_env_var, _parse_file_path, app
12 |
13 |
14 | @pytest.fixture
15 | def mock_config(tmp_path):
16 | """Create a mock Claude config file."""
17 | config = {"mcpServers": {}}
18 | config_file = tmp_path / "claude_desktop_config.json"
19 | config_file.write_text(json.dumps(config))
20 | return config_file
21 |
22 |
23 | @pytest.fixture
24 | def server_file(tmp_path):
25 | """Create a server file."""
26 | server_file = tmp_path / "server.py"
27 | server_file.write_text(
28 | """from fastmcp import FastMCP
29 | mcp = FastMCP("test")
30 | """
31 | )
32 | return server_file
33 |
34 |
35 | @pytest.fixture
36 | def mock_env_file(tmp_path):
37 | """Create a mock .env file."""
38 | env_file = tmp_path / ".env"
39 | env_file.write_text("FOO=bar\nBAZ=123")
40 | return env_file
41 |
42 |
43 | def test_parse_env_var():
44 | """Test parsing environment variables."""
45 | assert _parse_env_var("FOO=bar") == ("FOO", "bar")
46 | assert _parse_env_var("FOO=") == ("FOO", "")
47 | assert _parse_env_var("FOO=bar baz") == ("FOO", "bar baz")
48 | assert _parse_env_var("FOO = bar ") == ("FOO", "bar")
49 |
50 | with pytest.raises(SystemExit):
51 | _parse_env_var("invalid")
52 |
53 |
54 | @pytest.mark.parametrize(
55 | "args,expected_env",
56 | [
57 | # Basic env var
58 | (
59 | ["--env-var", "FOO=bar"],
60 | {"FOO": "bar"},
61 | ),
62 | # Multiple env vars
63 | (
64 | ["--env-var", "FOO=bar", "--env-var", "BAZ=123"],
65 | {"FOO": "bar", "BAZ": "123"},
66 | ),
67 | # Env var with spaces
68 | (
69 | ["--env-var", "FOO=bar baz"],
70 | {"FOO": "bar baz"},
71 | ),
72 | ],
73 | )
74 | def test_install_with_env_vars(mock_config, server_file, args, expected_env):
75 | """Test installing with environment variables."""
76 | runner = CliRunner()
77 |
78 | with patch("fastmcp.cli.claude.get_claude_config_path") as mock_config_path:
79 | mock_config_path.return_value = mock_config.parent
80 |
81 | result = runner.invoke(
82 | app,
83 | ["install", str(server_file)] + args,
84 | )
85 |
86 | assert result.exit_code == 0
87 |
88 | # Read the config file and check env vars
89 | config = json.loads(mock_config.read_text())
90 | assert "mcpServers" in config
91 | assert len(config["mcpServers"]) == 1
92 | server = next(iter(config["mcpServers"].values()))
93 | assert server["env"] == expected_env
94 |
95 |
96 | def test_parse_file_path_windows_drive():
97 | """Test parsing a Windows file path with a drive letter."""
98 | file_spec = r"C:\path\to\file.txt"
99 | with (
100 | patch("pathlib.Path.exists", return_value=True),
101 | patch("pathlib.Path.is_file", return_value=True),
102 | ):
103 | file_path, server_object = _parse_file_path(file_spec)
104 | assert file_path == Path(r"C:\path\to\file.txt").resolve()
105 | assert server_object is None
106 |
107 |
108 | def test_parse_file_path_with_object():
109 | """Test parsing a file path with an object specification."""
110 | file_spec = "/path/to/file.txt:object"
111 | with patch("sys.exit") as mock_exit:
112 | _parse_file_path(file_spec)
113 |
114 | # Check that sys.exit was called twice with code 1
115 | assert mock_exit.call_count == 2
116 | mock_exit.assert_has_calls([call(1), call(1)])
117 |
118 |
119 | def test_parse_file_path_windows_with_object():
120 | """Test parsing a Windows file path with an object specification."""
121 | file_spec = r"C:\path\to\file.txt:object"
122 | with (
123 | patch("pathlib.Path.exists", return_value=True),
124 | patch("pathlib.Path.is_file", return_value=True),
125 | ):
126 | file_path, server_object = _parse_file_path(file_spec)
127 | assert file_path == Path(r"C:\path\to\file.txt").resolve()
128 | assert server_object == "object"
129 |
130 |
131 | def test_install_with_env_file(mock_config, server_file, mock_env_file):
132 | """Test installing with environment variables from a file."""
133 | runner = CliRunner()
134 |
135 | with patch("fastmcp.cli.claude.get_claude_config_path") as mock_config_path:
136 | mock_config_path.return_value = mock_config.parent
137 |
138 | result = runner.invoke(
139 | app,
140 | ["install", str(server_file), "--env-file", str(mock_env_file)],
141 | )
142 |
143 | assert result.exit_code == 0
144 |
145 | # Read the config file and check env vars
146 | config = json.loads(mock_config.read_text())
147 | assert "mcpServers" in config
148 | assert len(config["mcpServers"]) == 1
149 | server = next(iter(config["mcpServers"].values()))
150 | assert server["env"] == {"FOO": "bar", "BAZ": "123"}
151 |
152 |
153 | def test_install_preserves_existing_env_vars(mock_config, server_file):
154 | """Test that installing preserves existing environment variables."""
155 | # Set up initial config with env vars
156 | config = {
157 | "mcpServers": {
158 | "test": {
159 | "command": "uv",
160 | "args": [
161 | "run",
162 | "--with",
163 | "fastmcp",
164 | "fastmcp",
165 | "run",
166 | str(server_file),
167 | ],
168 | "env": {"FOO": "bar", "BAZ": "123"},
169 | }
170 | }
171 | }
172 | mock_config.write_text(json.dumps(config))
173 |
174 | runner = CliRunner()
175 |
176 | with patch("fastmcp.cli.claude.get_claude_config_path") as mock_config_path:
177 | mock_config_path.return_value = mock_config.parent
178 |
179 | # Install with a new env var
180 | result = runner.invoke(
181 | app,
182 | ["install", str(server_file), "--env-var", "NEW=value"],
183 | )
184 |
185 | assert result.exit_code == 0
186 |
187 | # Read the config file and check env vars are preserved
188 | config = json.loads(mock_config.read_text())
189 | server = next(iter(config["mcpServers"].values()))
190 | assert server["env"] == {"FOO": "bar", "BAZ": "123", "NEW": "value"}
191 |
192 |
193 | def test_install_updates_existing_env_vars(mock_config, server_file):
194 | """Test that installing updates existing environment variables."""
195 | # Set up initial config with env vars
196 | config = {
197 | "mcpServers": {
198 | "test": {
199 | "command": "uv",
200 | "args": [
201 | "run",
202 | "--with",
203 | "fastmcp",
204 | "fastmcp",
205 | "run",
206 | str(server_file),
207 | ],
208 | "env": {"FOO": "bar", "BAZ": "123"},
209 | }
210 | }
211 | }
212 | mock_config.write_text(json.dumps(config))
213 |
214 | runner = CliRunner()
215 |
216 | with patch("fastmcp.cli.claude.get_claude_config_path") as mock_config_path:
217 | mock_config_path.return_value = mock_config.parent
218 |
219 | # Update an existing env var
220 | result = runner.invoke(
221 | app,
222 | ["install", str(server_file), "--env-var", "FOO=newvalue"],
223 | )
224 |
225 | assert result.exit_code == 0
226 |
227 | # Read the config file and check env var was updated
228 | config = json.loads(mock_config.read_text())
229 | server = next(iter(config["mcpServers"].values()))
230 | assert server["env"] == {"FOO": "newvalue", "BAZ": "123"}
231 |
232 |
233 | def test_server_dependencies(mock_config, server_file):
234 | """Test that server dependencies are correctly handled."""
235 | # Create a server file with dependencies
236 | server_file = server_file.parent / "server_with_deps.py"
237 | server_file.write_text(
238 | """from fastmcp import FastMCP
239 | mcp = FastMCP("test", dependencies=["pandas", "numpy"])
240 | """
241 | )
242 |
243 | runner = CliRunner()
244 |
245 | with patch("fastmcp.cli.claude.get_claude_config_path") as mock_config_path:
246 | mock_config_path.return_value = mock_config.parent
247 |
248 | result = runner.invoke(app, ["install", str(server_file)])
249 |
250 | assert result.exit_code == 0
251 |
252 | # Read the config file and check dependencies were added as --with args
253 | config = json.loads(mock_config.read_text())
254 | server = next(iter(config["mcpServers"].values()))
255 | assert "--with" in server["args"]
256 | assert "pandas" in server["args"]
257 | assert "numpy" in server["args"]
258 |
259 |
260 | def test_server_dependencies_empty(mock_config, server_file):
261 | """Test that server with no dependencies works correctly."""
262 | runner = CliRunner()
263 |
264 | with patch("fastmcp.cli.claude.get_claude_config_path") as mock_config_path:
265 | mock_config_path.return_value = mock_config.parent
266 |
267 | result = runner.invoke(app, ["install", str(server_file)])
268 |
269 | assert result.exit_code == 0
270 |
271 | # Read the config file and check only fastmcp is in --with args
272 | config = json.loads(mock_config.read_text())
273 | server = next(iter(config["mcpServers"].values()))
274 | assert server["args"].count("--with") == 1
275 | assert "fastmcp" in server["args"]
276 |
277 |
278 | def test_dev_with_dependencies(mock_config, server_file):
279 | """Test that dev command handles dependencies correctly."""
280 | server_file = server_file.parent / "server_with_deps.py"
281 | server_file.write_text(
282 | """from fastmcp import FastMCP
283 | mcp = FastMCP("test", dependencies=["pandas", "numpy"])
284 | """
285 | )
286 |
287 | runner = CliRunner()
288 |
289 | with patch("subprocess.run") as mock_run:
290 | mock_run.return_value.returncode = 0
291 | result = runner.invoke(app, ["dev", str(server_file)])
292 | assert result.exit_code == 0
293 |
294 | if sys.platform == "win32":
295 | # On Windows, expect two calls
296 | assert mock_run.call_count == 2
297 | assert mock_run.call_args_list[0] == call(
298 | ["npx.cmd", "--version"], check=True, capture_output=True, shell=True
299 | )
300 |
301 | # get the actual command and expected command without dependencies
302 | actual_cmd = mock_run.call_args_list[1][0][0]
303 | expected_start = [
304 | "npx.cmd",
305 | "@modelcontextprotocol/inspector",
306 | "uv",
307 | "run",
308 | "--with",
309 | "fastmcp",
310 | ]
311 | expected_end = ["fastmcp", "run", str(server_file)]
312 |
313 | # verify start and end of command
314 | assert actual_cmd[: len(expected_start)] == expected_start
315 | assert actual_cmd[-len(expected_end) :] == expected_end
316 |
317 | # verify dependencies are present (order-independent)
318 | deps_section = actual_cmd[len(expected_start) : -len(expected_end)]
319 | assert all(
320 | x in deps_section for x in ["--with", "numpy", "--with", "pandas"]
321 | )
322 |
323 | # Verify subprocess call kwargs, allowing for environment variables
324 | call_kwargs = mock_run.call_args_list[1][1]
325 | assert call_kwargs["check"] is True
326 | assert call_kwargs["shell"] is True
327 | assert isinstance(call_kwargs["env"], dict)
328 | else:
329 | # same verification for unix, just with different command prefix
330 | actual_cmd = mock_run.call_args_list[0][0][0]
331 | expected_start = [
332 | "npx",
333 | "@modelcontextprotocol/inspector",
334 | "uv",
335 | "run",
336 | "--with",
337 | "fastmcp",
338 | ]
339 | expected_end = ["fastmcp", "run", str(server_file)]
340 |
341 | assert actual_cmd[: len(expected_start)] == expected_start
342 | assert actual_cmd[-len(expected_end) :] == expected_end
343 |
344 | deps_section = actual_cmd[len(expected_start) : -len(expected_end)]
345 | assert all(
346 | x in deps_section for x in ["--with", "numpy", "--with", "pandas"]
347 | )
348 |
349 | # Verify subprocess call kwargs, allowing for environment variables
350 | call_kwargs = mock_run.call_args_list[0][1]
351 | assert call_kwargs["check"] is True
352 | assert call_kwargs["shell"] is False
353 | assert isinstance(call_kwargs["env"], dict)
354 |
355 |
356 | def test_run_with_dependencies(mock_config, server_file):
357 | """Test that run command does not handle dependencies."""
358 | # Create a server file with dependencies
359 | server_file = server_file.parent / "server_with_deps.py"
360 | server_file.write_text(
361 | """from fastmcp import FastMCP
362 | mcp = FastMCP("test", dependencies=["pandas", "numpy"])
363 |
364 | if __name__ == "__main__":
365 | mcp.run()
366 | """
367 | )
368 |
369 | runner = CliRunner()
370 |
371 | with patch("subprocess.run") as mock_run:
372 | result = runner.invoke(app, ["run", str(server_file)])
373 | assert result.exit_code == 0
374 |
375 | # Run command should not call subprocess.run
376 | mock_run.assert_not_called()
377 |
--------------------------------------------------------------------------------
/tests/test_func_metadata.py:
--------------------------------------------------------------------------------
1 | from typing import Annotated
2 |
3 | import annotated_types
4 | import pytest
5 | from pydantic import BaseModel, Field
6 |
7 | from fastmcp.utilities.func_metadata import func_metadata
8 |
9 |
10 | class SomeInputModelA(BaseModel):
11 | pass
12 |
13 |
14 | class SomeInputModelB(BaseModel):
15 | class InnerModel(BaseModel):
16 | x: int
17 |
18 | how_many_shrimp: Annotated[int, Field(description="How many shrimp in the tank???")]
19 | ok: InnerModel
20 | y: None
21 |
22 |
23 | def complex_arguments_fn(
24 | an_int: int,
25 | must_be_none: None,
26 | must_be_none_dumb_annotation: Annotated[None, "blah"],
27 | list_of_ints: list[int],
28 | # list[str] | str is an interesting case because if it comes in as JSON like
29 | # "[\"a\", \"b\"]" then it will be naively parsed as a string.
30 | list_str_or_str: list[str] | str,
31 | an_int_annotated_with_field: Annotated[
32 | int, Field(description="An int with a field")
33 | ],
34 | an_int_annotated_with_field_and_others: Annotated[
35 | int,
36 | str, # Should be ignored, really
37 | Field(description="An int with a field"),
38 | annotated_types.Gt(1),
39 | ],
40 | an_int_annotated_with_junk: Annotated[
41 | int,
42 | "123",
43 | 456,
44 | ],
45 | field_with_default_via_field_annotation_before_nondefault_arg: Annotated[
46 | int, Field(1)
47 | ],
48 | unannotated,
49 | my_model_a: SomeInputModelA,
50 | my_model_a_forward_ref: "SomeInputModelA",
51 | my_model_b: SomeInputModelB,
52 | an_int_annotated_with_field_default: Annotated[
53 | int,
54 | Field(1, description="An int with a field"),
55 | ],
56 | unannotated_with_default=5,
57 | my_model_a_with_default: SomeInputModelA = SomeInputModelA(), # noqa: B008
58 | an_int_with_default: int = 1,
59 | must_be_none_with_default: None = None,
60 | an_int_with_equals_field: int = Field(1, ge=0),
61 | int_annotated_with_default: Annotated[int, Field(description="hey")] = 5,
62 | ) -> str:
63 | _ = (
64 | an_int,
65 | must_be_none,
66 | must_be_none_dumb_annotation,
67 | list_of_ints,
68 | list_str_or_str,
69 | an_int_annotated_with_field,
70 | an_int_annotated_with_field_and_others,
71 | an_int_annotated_with_junk,
72 | field_with_default_via_field_annotation_before_nondefault_arg,
73 | unannotated,
74 | an_int_annotated_with_field_default,
75 | unannotated_with_default,
76 | my_model_a,
77 | my_model_a_forward_ref,
78 | my_model_b,
79 | my_model_a_with_default,
80 | an_int_with_default,
81 | must_be_none_with_default,
82 | an_int_with_equals_field,
83 | int_annotated_with_default,
84 | )
85 | return "ok!"
86 |
87 |
88 | async def test_complex_function_runtime_arg_validation_non_json():
89 | """Test that basic non-JSON arguments are validated correctly"""
90 | meta = func_metadata(complex_arguments_fn)
91 |
92 | # Test with minimum required arguments
93 | result = await meta.call_fn_with_arg_validation(
94 | complex_arguments_fn,
95 | fn_is_async=False,
96 | arguments_to_validate={
97 | "an_int": 1,
98 | "must_be_none": None,
99 | "must_be_none_dumb_annotation": None,
100 | "list_of_ints": [1, 2, 3],
101 | "list_str_or_str": "hello",
102 | "an_int_annotated_with_field": 42,
103 | "an_int_annotated_with_field_and_others": 5,
104 | "an_int_annotated_with_junk": 100,
105 | "unannotated": "test",
106 | "my_model_a": {},
107 | "my_model_a_forward_ref": {},
108 | "my_model_b": {"how_many_shrimp": 5, "ok": {"x": 1}, "y": None},
109 | },
110 | arguments_to_pass_directly=None,
111 | )
112 | assert result == "ok!"
113 |
114 | # Test with invalid types
115 | with pytest.raises(ValueError):
116 | await meta.call_fn_with_arg_validation(
117 | complex_arguments_fn,
118 | fn_is_async=False,
119 | arguments_to_validate={"an_int": "not an int"},
120 | arguments_to_pass_directly=None,
121 | )
122 |
123 |
124 | async def test_complex_function_runtime_arg_validation_with_json():
125 | """Test that JSON string arguments are parsed and validated correctly"""
126 | meta = func_metadata(complex_arguments_fn)
127 |
128 | result = await meta.call_fn_with_arg_validation(
129 | complex_arguments_fn,
130 | fn_is_async=False,
131 | arguments_to_validate={
132 | "an_int": 1,
133 | "must_be_none": None,
134 | "must_be_none_dumb_annotation": None,
135 | "list_of_ints": "[1, 2, 3]", # JSON string
136 | "list_str_or_str": '["a", "b", "c"]', # JSON string
137 | "an_int_annotated_with_field": 42,
138 | "an_int_annotated_with_field_and_others": "5", # JSON string
139 | "an_int_annotated_with_junk": 100,
140 | "unannotated": "test",
141 | "my_model_a": "{}", # JSON string
142 | "my_model_a_forward_ref": "{}", # JSON string
143 | "my_model_b": '{"how_many_shrimp": 5, "ok": {"x": 1}, "y": null}', # JSON string
144 | },
145 | arguments_to_pass_directly=None,
146 | )
147 | assert result == "ok!"
148 |
149 |
150 | def test_str_vs_list_str():
151 | """Test handling of string vs list[str] type annotations.
152 |
153 | This is tricky as '"hello"' can be parsed as a JSON string or a Python string.
154 | We want to make sure it's kept as a python string.
155 | """
156 |
157 | def func_with_str_types(str_or_list: str | list[str]):
158 | return str_or_list
159 |
160 | meta = func_metadata(func_with_str_types)
161 |
162 | # Test string input for union type
163 | result = meta.pre_parse_json({"str_or_list": "hello"})
164 | assert result["str_or_list"] == "hello"
165 |
166 | # Test string input that contains valid JSON for union type
167 | # We want to see here that the JSON-vali string is NOT parsed as JSON, but rather
168 | # kept as a raw string
169 | result = meta.pre_parse_json({"str_or_list": '"hello"'})
170 | assert result["str_or_list"] == '"hello"'
171 |
172 | # Test list input for union type
173 | result = meta.pre_parse_json({"str_or_list": '["hello", "world"]'})
174 | assert result["str_or_list"] == ["hello", "world"]
175 |
176 |
177 | def test_str_vs_int():
178 | """
179 | Test that string values are kept as strings even when they contain numbers,
180 | while numbers are parsed correctly.
181 | """
182 |
183 | def func_with_str_and_int(a: str, b: int):
184 | return a
185 |
186 | meta = func_metadata(func_with_str_and_int)
187 | result = meta.pre_parse_json({"a": "123", "b": 123})
188 | assert result["a"] == "123"
189 | assert result["b"] == 123
190 |
191 |
192 | def test_skip_names():
193 | """Test that skipped parameters are not included in the model"""
194 |
195 | def func_with_many_params(
196 | keep_this: int, skip_this: str, also_keep: float, also_skip: bool
197 | ):
198 | return keep_this, skip_this, also_keep, also_skip
199 |
200 | # Skip some parameters
201 | meta = func_metadata(func_with_many_params, skip_names=["skip_this", "also_skip"])
202 |
203 | # Check model fields
204 | assert "keep_this" in meta.arg_model.model_fields
205 | assert "also_keep" in meta.arg_model.model_fields
206 | assert "skip_this" not in meta.arg_model.model_fields
207 | assert "also_skip" not in meta.arg_model.model_fields
208 |
209 | # Validate that we can call with only non-skipped parameters
210 | model: BaseModel = meta.arg_model.model_validate({"keep_this": 1, "also_keep": 2.5}) # type: ignore
211 | assert model.keep_this == 1 # type: ignore
212 | assert model.also_keep == 2.5 # type: ignore
213 |
214 |
215 | async def test_lambda_function():
216 | """Test lambda function schema and validation"""
217 | fn = lambda x, y=5: x # noqa: E731
218 | meta = func_metadata(lambda x, y=5: x)
219 |
220 | # Test schema
221 | assert meta.arg_model.model_json_schema() == {
222 | "properties": {
223 | "x": {"title": "x", "type": "string"},
224 | "y": {"default": 5, "title": "y", "type": "string"},
225 | },
226 | "required": ["x"],
227 | "title": "<lambda>Arguments",
228 | "type": "object",
229 | }
230 |
231 | async def check_call(args):
232 | return await meta.call_fn_with_arg_validation(
233 | fn,
234 | fn_is_async=False,
235 | arguments_to_validate=args,
236 | arguments_to_pass_directly=None,
237 | )
238 |
239 | # Basic calls
240 | assert await check_call({"x": "hello"}) == "hello"
241 | assert await check_call({"x": "hello", "y": "world"}) == "hello"
242 | assert await check_call({"x": '"hello"'}) == '"hello"'
243 |
244 | # Missing required arg
245 | with pytest.raises(ValueError):
246 | await check_call({"y": "world"})
247 |
248 |
249 | def test_complex_function_json_schema():
250 | meta = func_metadata(complex_arguments_fn)
251 | assert meta.arg_model.model_json_schema() == {
252 | "$defs": {
253 | "InnerModel": {
254 | "properties": {"x": {"title": "X", "type": "integer"}},
255 | "required": ["x"],
256 | "title": "InnerModel",
257 | "type": "object",
258 | },
259 | "SomeInputModelA": {
260 | "properties": {},
261 | "title": "SomeInputModelA",
262 | "type": "object",
263 | },
264 | "SomeInputModelB": {
265 | "properties": {
266 | "how_many_shrimp": {
267 | "description": "How many shrimp in the tank???",
268 | "title": "How Many Shrimp",
269 | "type": "integer",
270 | },
271 | "ok": {"$ref": "#/$defs/InnerModel"},
272 | "y": {"title": "Y", "type": "null"},
273 | },
274 | "required": ["how_many_shrimp", "ok", "y"],
275 | "title": "SomeInputModelB",
276 | "type": "object",
277 | },
278 | },
279 | "properties": {
280 | "an_int": {"title": "An Int", "type": "integer"},
281 | "must_be_none": {"title": "Must Be None", "type": "null"},
282 | "must_be_none_dumb_annotation": {
283 | "title": "Must Be None Dumb Annotation",
284 | "type": "null",
285 | },
286 | "list_of_ints": {
287 | "items": {"type": "integer"},
288 | "title": "List Of Ints",
289 | "type": "array",
290 | },
291 | "list_str_or_str": {
292 | "anyOf": [
293 | {"items": {"type": "string"}, "type": "array"},
294 | {"type": "string"},
295 | ],
296 | "title": "List Str Or Str",
297 | },
298 | "an_int_annotated_with_field": {
299 | "description": "An int with a field",
300 | "title": "An Int Annotated With Field",
301 | "type": "integer",
302 | },
303 | "an_int_annotated_with_field_and_others": {
304 | "description": "An int with a field",
305 | "exclusiveMinimum": 1,
306 | "title": "An Int Annotated With Field And Others",
307 | "type": "integer",
308 | },
309 | "an_int_annotated_with_junk": {
310 | "title": "An Int Annotated With Junk",
311 | "type": "integer",
312 | },
313 | "field_with_default_via_field_annotation_before_nondefault_arg": {
314 | "default": 1,
315 | "title": "Field With Default Via Field Annotation Before Nondefault Arg",
316 | "type": "integer",
317 | },
318 | "unannotated": {"title": "unannotated", "type": "string"},
319 | "my_model_a": {"$ref": "#/$defs/SomeInputModelA"},
320 | "my_model_a_forward_ref": {"$ref": "#/$defs/SomeInputModelA"},
321 | "my_model_b": {"$ref": "#/$defs/SomeInputModelB"},
322 | "an_int_annotated_with_field_default": {
323 | "default": 1,
324 | "description": "An int with a field",
325 | "title": "An Int Annotated With Field Default",
326 | "type": "integer",
327 | },
328 | "unannotated_with_default": {
329 | "default": 5,
330 | "title": "unannotated_with_default",
331 | "type": "string",
332 | },
333 | "my_model_a_with_default": {
334 | "$ref": "#/$defs/SomeInputModelA",
335 | "default": {},
336 | },
337 | "an_int_with_default": {
338 | "default": 1,
339 | "title": "An Int With Default",
340 | "type": "integer",
341 | },
342 | "must_be_none_with_default": {
343 | "default": None,
344 | "title": "Must Be None With Default",
345 | "type": "null",
346 | },
347 | "an_int_with_equals_field": {
348 | "default": 1,
349 | "minimum": 0,
350 | "title": "An Int With Equals Field",
351 | "type": "integer",
352 | },
353 | "int_annotated_with_default": {
354 | "default": 5,
355 | "description": "hey",
356 | "title": "Int Annotated With Default",
357 | "type": "integer",
358 | },
359 | },
360 | "required": [
361 | "an_int",
362 | "must_be_none",
363 | "must_be_none_dumb_annotation",
364 | "list_of_ints",
365 | "list_str_or_str",
366 | "an_int_annotated_with_field",
367 | "an_int_annotated_with_field_and_others",
368 | "an_int_annotated_with_junk",
369 | "unannotated",
370 | "my_model_a",
371 | "my_model_a_forward_ref",
372 | "my_model_b",
373 | ],
374 | "title": "complex_arguments_fnArguments",
375 | "type": "object",
376 | }
377 |
--------------------------------------------------------------------------------
/tests/test_tool_manager.py:
--------------------------------------------------------------------------------
1 | import logging
2 | from typing import Optional
3 |
4 | import pytest
5 | from pydantic import BaseModel
6 | import json
7 | from fastmcp.exceptions import ToolError
8 | from fastmcp.tools import ToolManager
9 |
10 |
11 | class TestAddTools:
12 | def test_basic_function(self):
13 | """Test registering and running a basic function."""
14 |
15 | def add(a: int, b: int) -> int:
16 | """Add two numbers."""
17 | return a + b
18 |
19 | manager = ToolManager()
20 | manager.add_tool(add)
21 |
22 | tool = manager.get_tool("add")
23 | assert tool is not None
24 | assert tool.name == "add"
25 | assert tool.description == "Add two numbers."
26 | assert tool.is_async is False
27 | assert tool.parameters["properties"]["a"]["type"] == "integer"
28 | assert tool.parameters["properties"]["b"]["type"] == "integer"
29 |
30 | async def test_async_function(self):
31 | """Test registering and running an async function."""
32 |
33 | async def fetch_data(url: str) -> str:
34 | """Fetch data from URL."""
35 | return f"Data from {url}"
36 |
37 | manager = ToolManager()
38 | manager.add_tool(fetch_data)
39 |
40 | tool = manager.get_tool("fetch_data")
41 | assert tool is not None
42 | assert tool.name == "fetch_data"
43 | assert tool.description == "Fetch data from URL."
44 | assert tool.is_async is True
45 | assert tool.parameters["properties"]["url"]["type"] == "string"
46 |
47 | def test_pydantic_model_function(self):
48 | """Test registering a function that takes a Pydantic model."""
49 |
50 | class UserInput(BaseModel):
51 | name: str
52 | age: int
53 |
54 | def create_user(user: UserInput, flag: bool) -> dict:
55 | """Create a new user."""
56 | return {"id": 1, **user.model_dump()}
57 |
58 | manager = ToolManager()
59 | manager.add_tool(create_user)
60 |
61 | tool = manager.get_tool("create_user")
62 | assert tool is not None
63 | assert tool.name == "create_user"
64 | assert tool.description == "Create a new user."
65 | assert tool.is_async is False
66 | assert "name" in tool.parameters["$defs"]["UserInput"]["properties"]
67 | assert "age" in tool.parameters["$defs"]["UserInput"]["properties"]
68 | assert "flag" in tool.parameters["properties"]
69 |
70 | def test_add_invalid_tool(self):
71 | manager = ToolManager()
72 | with pytest.raises(AttributeError):
73 | manager.add_tool(1) # type: ignore
74 |
75 | def test_add_lambda(self):
76 | manager = ToolManager()
77 | tool = manager.add_tool(lambda x: x, name="my_tool")
78 | assert tool.name == "my_tool"
79 |
80 | def test_add_lambda_with_no_name(self):
81 | manager = ToolManager()
82 | with pytest.raises(
83 | ValueError, match="You must provide a name for lambda functions"
84 | ):
85 | manager.add_tool(lambda x: x)
86 |
87 | def test_warn_on_duplicate_tools(self, caplog):
88 | """Test warning on duplicate tools."""
89 |
90 | def f(x: int) -> int:
91 | return x
92 |
93 | manager = ToolManager()
94 | manager.add_tool(f)
95 | with caplog.at_level(logging.WARNING):
96 | manager.add_tool(f)
97 | assert "Tool already exists: f" in caplog.text
98 |
99 | def test_disable_warn_on_duplicate_tools(self, caplog):
100 | """Test disabling warning on duplicate tools."""
101 |
102 | def f(x: int) -> int:
103 | return x
104 |
105 | manager = ToolManager()
106 | manager.add_tool(f)
107 | manager.warn_on_duplicate_tools = False
108 | with caplog.at_level(logging.WARNING):
109 | manager.add_tool(f)
110 | assert "Tool already exists: f" not in caplog.text
111 |
112 |
113 | class TestCallTools:
114 | async def test_call_tool(self):
115 | def add(a: int, b: int) -> int:
116 | """Add two numbers."""
117 | return a + b
118 |
119 | manager = ToolManager()
120 | manager.add_tool(add)
121 | result = await manager.call_tool("add", {"a": 1, "b": 2})
122 | assert result == 3
123 |
124 | async def test_call_async_tool(self):
125 | async def double(n: int) -> int:
126 | """Double a number."""
127 | return n * 2
128 |
129 | manager = ToolManager()
130 | manager.add_tool(double)
131 | result = await manager.call_tool("double", {"n": 5})
132 | assert result == 10
133 |
134 | async def test_call_tool_with_default_args(self):
135 | def add(a: int, b: int = 1) -> int:
136 | """Add two numbers."""
137 | return a + b
138 |
139 | manager = ToolManager()
140 | manager.add_tool(add)
141 | result = await manager.call_tool("add", {"a": 1})
142 | assert result == 2
143 |
144 | async def test_call_tool_with_missing_args(self):
145 | def add(a: int, b: int) -> int:
146 | """Add two numbers."""
147 | return a + b
148 |
149 | manager = ToolManager()
150 | manager.add_tool(add)
151 | with pytest.raises(ToolError):
152 | await manager.call_tool("add", {"a": 1})
153 |
154 | async def test_call_unknown_tool(self):
155 | manager = ToolManager()
156 | with pytest.raises(ToolError):
157 | await manager.call_tool("unknown", {"a": 1})
158 |
159 | async def test_call_tool_with_list_int_input(self):
160 | def sum_vals(vals: list[int]) -> int:
161 | return sum(vals)
162 |
163 | manager = ToolManager()
164 | manager.add_tool(sum_vals)
165 | # Try both with plain list and with JSON list
166 | result = await manager.call_tool("sum_vals", {"vals": "[1, 2, 3]"})
167 | assert result == 6
168 | result = await manager.call_tool("sum_vals", {"vals": [1, 2, 3]})
169 | assert result == 6
170 |
171 | async def test_call_tool_with_list_str_or_str_input(self):
172 | def concat_strs(vals: list[str] | str) -> str:
173 | return vals if isinstance(vals, str) else "".join(vals)
174 |
175 | manager = ToolManager()
176 | manager.add_tool(concat_strs)
177 | # Try both with plain python object and with JSON list
178 | result = await manager.call_tool("concat_strs", {"vals": ["a", "b", "c"]})
179 | assert result == "abc"
180 | result = await manager.call_tool("concat_strs", {"vals": '["a", "b", "c"]'})
181 | assert result == "abc"
182 | result = await manager.call_tool("concat_strs", {"vals": "a"})
183 | assert result == "a"
184 | result = await manager.call_tool("concat_strs", {"vals": '"a"'})
185 | assert result == '"a"'
186 |
187 | async def test_call_tool_with_complex_model(self):
188 | from fastmcp import Context
189 |
190 | class MyShrimpTank(BaseModel):
191 | class Shrimp(BaseModel):
192 | name: str
193 |
194 | shrimp: list[Shrimp]
195 | x: None
196 |
197 | def name_shrimp(tank: MyShrimpTank, ctx: Context) -> list[str]:
198 | return [x.name for x in tank.shrimp]
199 |
200 | manager = ToolManager()
201 | manager.add_tool(name_shrimp)
202 | result = await manager.call_tool(
203 | "name_shrimp",
204 | {"tank": {"x": None, "shrimp": [{"name": "rex"}, {"name": "gertrude"}]}},
205 | )
206 | assert result == ["rex", "gertrude"]
207 | result = await manager.call_tool(
208 | "name_shrimp",
209 | {"tank": '{"x": null, "shrimp": [{"name": "rex"}, {"name": "gertrude"}]}'},
210 | )
211 | assert result == ["rex", "gertrude"]
212 |
213 |
214 | class TestToolSchema:
215 | async def test_context_arg_excluded_from_schema(self):
216 | from fastmcp import Context
217 |
218 | def something(a: int, ctx: Context) -> int:
219 | return a
220 |
221 | manager = ToolManager()
222 | tool = manager.add_tool(something)
223 | assert "ctx" not in json.dumps(tool.parameters)
224 | assert "Context" not in json.dumps(tool.parameters)
225 | assert "ctx" not in tool.fn_metadata.arg_model.model_fields
226 |
227 |
228 | class TestContextHandling:
229 | """Test context handling in the tool manager."""
230 |
231 | def test_context_parameter_detection(self):
232 | """Test that context parameters are properly detected in Tool.from_function()."""
233 | from fastmcp import Context
234 |
235 | def tool_with_context(x: int, ctx: Context) -> str:
236 | return str(x)
237 |
238 | manager = ToolManager()
239 | tool = manager.add_tool(tool_with_context)
240 | assert tool.context_kwarg == "ctx"
241 |
242 | def tool_without_context(x: int) -> str:
243 | return str(x)
244 |
245 | tool = manager.add_tool(tool_without_context)
246 | assert tool.context_kwarg is None
247 |
248 | async def test_context_injection(self):
249 | """Test that context is properly injected during tool execution."""
250 | from fastmcp import Context, FastMCP
251 |
252 | def tool_with_context(x: int, ctx: Context) -> str:
253 | assert isinstance(ctx, Context)
254 | return str(x)
255 |
256 | manager = ToolManager()
257 | manager.add_tool(tool_with_context)
258 |
259 | mcp = FastMCP()
260 | ctx = mcp.get_context()
261 | result = await manager.call_tool("tool_with_context", {"x": 42}, context=ctx)
262 | assert result == "42"
263 |
264 | async def test_context_injection_async(self):
265 | """Test that context is properly injected in async tools."""
266 | from fastmcp import Context, FastMCP
267 |
268 | async def async_tool(x: int, ctx: Context) -> str:
269 | assert isinstance(ctx, Context)
270 | return str(x)
271 |
272 | manager = ToolManager()
273 | manager.add_tool(async_tool)
274 |
275 | mcp = FastMCP()
276 | ctx = mcp.get_context()
277 | result = await manager.call_tool("async_tool", {"x": 42}, context=ctx)
278 | assert result == "42"
279 |
280 | async def test_context_optional(self):
281 | """Test that context is optional when calling tools."""
282 | from fastmcp import Context
283 |
284 | def tool_with_context(x: int, ctx: Optional[Context] = None) -> str:
285 | return str(x)
286 |
287 | manager = ToolManager()
288 | manager.add_tool(tool_with_context)
289 | # Should not raise an error when context is not provided
290 | result = await manager.call_tool("tool_with_context", {"x": 42})
291 | assert result == "42"
292 |
293 | async def test_context_error_handling(self):
294 | """Test error handling when context injection fails."""
295 | from fastmcp import Context, FastMCP
296 |
297 | def tool_with_context(x: int, ctx: Context) -> str:
298 | raise ValueError("Test error")
299 |
300 | manager = ToolManager()
301 | manager.add_tool(tool_with_context)
302 |
303 | mcp = FastMCP()
304 | ctx = mcp.get_context()
305 | with pytest.raises(ToolError, match="Error executing tool tool_with_context"):
306 | await manager.call_tool("tool_with_context", {"x": 42}, context=ctx)
307 |
--------------------------------------------------------------------------------