Alright, folks. Here we go.
In this simple example I’ve whipped up to show you what’s what, the main idea is to have a local MCP server (that’s the server.py file) which basically just wraps the duckduckgo_search library. As a solid best practice, you’ll see a class used to standardize the tool’s output. You should definitely make it a habit when you’re building tools for LLMs.
For the client side, we’ve got a pretty straightforward crew that uses these tools. First off, it hits the internet for a search, and then it digs through the content of each result it finds. In this example, I’m bringing together both an MCP tool (that’s our server) and a traditional, out-of-the-box CrewAI tool. This is to show you how you can mix and match different kinds of tools to open up a whole bunch of possibilities.
And for a little something new, you’ll see I’m using a more souped-up version of ScrapeWebsiteTool. You can, and absolutely should, grab it here and save it in the same directory as your other files. This tool might even end up being a proposal to replace CrewAI’s original ScrapeWebsiteTool, 'cause it gives you options to limit context window usage (the original one just sends back the entire page content, no matter how huge it is). By the end of this week, @tonykipkemboi and I are gonna decide if it’s gonna make it into the next CrewAI version. For now, though, go ahead and download it to get a feel for it and give it a whirl.
One last new tidbit is the Task.markdown attribute, which was rolled out not too long ago.
To install the libraries you’ll need:
pip install --upgrade crewai crewai-tools[mcp] duckduckgo-search
Directory structure:
crewai_mcp_tool_example/
├── crew.py
├── server.py
└── versatile_scrape_website_tool.py
server.py file:
from typing import List, Optional, Dict
from pydantic import BaseModel, Field, ConfigDict
from mcp.server.fastmcp import FastMCP
from duckduckgo_search import DDGS
from duckduckgo_search.exceptions import DuckDuckGoSearchException
class DuckDuckGoSearchResultItem(BaseModel):
"""Represents a single search result item from DuckDuckGo."""
href: str
title: str
body: str
model_config = ConfigDict(extra="forbid")
class DuckDuckGoSearchToolOutput(BaseModel):
"""Standardized output for the duckduckgo_text_search tool."""
query_used: Optional[str] = Field(
default=None, description="The query string used for the search."
)
search_results: List[DuckDuckGoSearchResultItem] = Field(
default_factory=list,
description=(
"A list of search results. Empty if no results were found "
"or an error occurred."
),
)
error_message: Optional[str] = Field(
default=None,
description=(
"An error message if the search process failed, "
"no query was provided, or no results were found."
),
)
model_config = ConfigDict(extra="forbid", validate_assignment=True)
def to_llm_response(self) -> str:
"""Converts the output to a JSON string for the LLM."""
return self.model_dump_json(exclude_none=True, indent=2)
mcp_service = FastMCP(
name="DuckDuckGoSearchService",
description="Provides DuckDuckGo web search functionality.",
)
@mcp_service.tool()
def duckduckgo_text_search(query: Optional[str] = None) -> str:
"""
Performs a DuckDuckGo web search for the given query.
If no query is provided, an error message is returned.
The results are returned as a JSON string based on the
DuckDuckGoSearchToolOutput model.
Args:
query: The search query string. This is optional. If not
provided or empty, an error will be indicated in the
output.
Returns:
A JSON string representing the search outcome, including
results or an error message.
"""
if not query or not query.strip():
error_msg = "Query parameter is missing or empty."
print(f"[MCP Server] Error: {error_msg}")
output = DuckDuckGoSearchToolOutput(
query_used=query if query else None, error_message=error_msg
)
return output.to_llm_response()
print(f"[MCP Server] Received text_search request: query='{query}'")
# Hardcoded parameters for simplicity
region: str = "wt-wt"
safesearch: str = "moderate"
timelimit: Optional[str] = None
max_results: int = 5 # Fetch top 5 results
try:
ddgs_client = DDGS(timeout=10)
# ddgs.text returns List[Dict] or None
raw_results: Optional[List[Dict[str, str]]] = ddgs_client.text(
keywords=query,
region=region,
safesearch=safesearch,
timelimit=timelimit,
max_results=max_results,
)
if raw_results:
parsed_results = [
DuckDuckGoSearchResultItem(**item) for item in raw_results
]
output = DuckDuckGoSearchToolOutput(
query_used=query, search_results=parsed_results
)
print(
f"[MCP Server] Found {len(parsed_results)} results "
f"for query: '{query}'"
)
else:
# No results found by DDGS, or DDGS returned None
msg = "No results found for the query."
print(f"[MCP Server] {msg} Query: '{query}'")
output = DuckDuckGoSearchToolOutput(
query_used=query,
error_message=msg,
# search_results is an empty list by default
)
except DuckDuckGoSearchException as e:
error_msg = f"DuckDuckGo search API error: {str(e)}"
print(f"[MCP Server] Error: {error_msg}")
output = DuckDuckGoSearchToolOutput(
query_used=query, error_message=error_msg
)
except Exception as e:
# Catch any other unexpected errors during the process
error_msg = (
f"An unexpected error occurred in "
f"duckduckgo_text_search: {str(e)}"
)
print(f"[MCP Server] Error: {error_msg}")
output = DuckDuckGoSearchToolOutput(
query_used=query, error_message=error_msg
)
return output.to_llm_response()
if __name__ == "__main__":
print(
"Starting DuckDuckGo Search MCP Server via Stdio...\n"
"This server will listen for MCP messages on stdin/stdout."
)
mcp_service.run(transport="stdio")
crew.py file:
from crewai import Agent, Task, Crew, LLM, Process
from crewai_tools import MCPServerAdapter
from mcp import StdioServerParameters
from versatile_scrape_website_tool import VersatileScrapeWebsiteTool as ScrapeWebsiteTool
import os
import warnings
from pydantic.warnings import PydanticDeprecatedSince20
warnings.filterwarnings("ignore", category=PydanticDeprecatedSince20)
os.environ["GEMINI_API_KEY"] = "<YOUR_GEMINI_API_KEY>"
ddgs_stdio_server_params = StdioServerParameters(
command="python3",
args=["server.py"],
env={"UV_PYTHON": "3.12", **os.environ},
)
try:
ddgs_mcp_server_adapter = MCPServerAdapter(
serverparams=ddgs_stdio_server_params
)
ddgs_mcp_tools = ddgs_mcp_server_adapter.tools
scrape_website_tool = ScrapeWebsiteTool(
retrieval_mode="random_chunks",
max_chars=6000
)
all_tools = [*ddgs_mcp_tools, scrape_website_tool]
gemini_llm = LLM(
model="gemini/gemini-2.5-flash-preview-04-17",
temperature=0.7,
timeout=30,
)
study_guide_agent = Agent(
role="Expert Study Guide Creator",
goal=(
"To research topics online and create structured learning "
"paths that guide students from basic to advanced concepts."
),
backstory=(
"You are a seasoned educator with expertise in curriculum "
"design. You excel at breaking down complex topics into "
"digestible learning steps, finding the best online resources, "
"and creating progressive study plans that build knowledge "
"systematically from fundamentals to advanced concepts."
),
tools=all_tools,
llm=gemini_llm,
allow_delegation=False,
verbose=True,
)
learning_path_task = Task(
description=(
"The user has interest in better learning about:\n\n"
"{user_interest}\n\n"
"Using EXCLUSIVELY the available tools, you must:\n\n"
"1. Search the internet for interesting sites about it\n"
"2. Scrape EVERY searched site to evaluate the knowledge "
"offered there\n"
"3. Create a study plan for the user with 3 to 5 steps "
"for learning, from least to most complex, including a "
"brief summary of what they'll learn at each step and "
"recommended URLs from those you visited."
),
expected_output=(
"A comprehensive study plan with 3-5 progressive learning "
"steps, each containing: step title, learning objectives, "
"content summary, and recommended website URL. Steps should "
"flow from basic concepts to advanced applications."
),
markdown=True,
agent=study_guide_agent,
)
study_guide_crew = Crew(
agents=[study_guide_agent],
tasks=[learning_path_task],
process=Process.sequential,
verbose=True,
)
user_interest = "Quantum Physics for Beginners"
learning_result = study_guide_crew.kickoff(
inputs={"user_interest": user_interest}
)
print(learning_result.raw)
finally:
ddgs_mcp_server_adapter.stop()
To get it running, just pop open your terminal and run python3 crew.py – it’ll take care of firing up the server. Hope this helps y’all out!