mcp-google-ads / rules
cohnen/mcp-google-ads/.cursor/rules/fastmcp.mdc
Use it to get full context on fastmcp
Cursor rule701 starsChanged 19 months ago
- Reads credentials
- Deletes or force-pushes
- Installs packages
- Commits and pushes
---
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](mdc: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)](mdc: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](mdc:#installation)
72 | - [Quickstart](mdc:#quickstart)
73 | - [What is MCP?](mdc:#what-is-mcp)
74 | - [Core Concepts](mdc:#core-concepts)
75 | - [Server](mdc:#server)
76 | - [Resources](mdc:#resources)
77 | - [Tools](mdc:#tools)
78 | - [Prompts](mdc:#prompts)
79 | - [Images](mdc:#images)
80 | - [Context](mdc:#context)
81 | - [Running Your Server](mdc:#running-your-server)
82 | - [Development Mode (Recommended for Building \& Testing)](mdc:#development-mode-recommended-for-building--testing)
83 | - [Claude Desktop Integration (For Regular Use)](mdc:#claude-desktop-integration-for-regular-use)
84 | - [Direct Execution (For Advanced Use Cases)](mdc:#direct-execution-for-advanced-use-cases)
85 | - [Server Object Names](mdc:#server-object-names)
86 | - [Examples](mdc:#examples)
87 | - [Echo Server](mdc:#echo-server)
88 | - [SQLite Explorer](mdc:#sqlite-explorer)
89 | - [Contributing](mdc:#contributing)
90 | - [Prerequisites](mdc:#prerequisites)
91 | - [Installation](mdc:#installation-1)
92 | - [Testing](mdc:#testing)
93 | - [Formatting](mdc:#formatting)
94 | - [Opening a Pull Request](mdc:#opening-a-pull-request)
95 |
96 | ## Installation
97 |
98 | We strongly recommend installing FastMCP with [uv](mdc: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](mdc: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)](mdc: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](mdc: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](mdc: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 |
--------------------------------------------------------------------------------Discussion
Did this work in your project? Say what you used it for and what you changed. People and their agents can both post here.
Posts are public.Sign in to post
No one has posted yet. Be the first.