diff --git a/.github/PULL_REQUEST_TEMPLATE.md b/.github/PULL_REQUEST_TEMPLATE.md new file mode 100644 index 0000000..c4f06a3 --- /dev/null +++ b/.github/PULL_REQUEST_TEMPLATE.md @@ -0,0 +1,35 @@ +## Description + + + +## Related Issues + + + +## Type of Change + + + +Bug fix +New Feature +Breaking change +Documentation update +Other (please describe): + +## Testing + +How have you tested the change? + +- [ ] I ran `hatch run prepare` +- [ ] All tests pass locally + +## Checklist +- [ ] I have read the CONTRIBUTING document +- [ ] I have added any necessary tests that prove my fix is effective or my feature works +- [ ] I have updated the documentation accordingly +- [ ] My changes generate no new warnings +- [ ] Any dependent changes have been merged and published + +---- + +By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice. \ No newline at end of file diff --git a/.github/dependabot.yml b/.github/dependabot.yml index 514c791..addb4fc 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -7,10 +7,14 @@ updates: open-pull-requests-limit: 100 commit-message: prefix: ci + groups: + dev-dependencies: + patterns: + - "pytest" - package-ecosystem: "github-actions" directory: "/" schedule: interval: "daily" open-pull-requests-limit: 100 commit-message: - prefix: ci + prefix: ci \ No newline at end of file diff --git a/.github/workflows/pr-and-push.yml b/.github/workflows/pr-and-push.yml index 456b8a2..52686e5 100644 --- a/.github/workflows/pr-and-push.yml +++ b/.github/workflows/pr-and-push.yml @@ -3,7 +3,7 @@ name: Pull Request and Push Action on: pull_request: # Safer than pull_request_target for untrusted code branches: [ main ] - types: [opened, synchronize, reopened, ready_for_review, review_requested, review_request_removed] + types: [opened, synchronize, reopened, ready_for_review] push: branches: [ main ] # Also run on direct pushes to main concurrency: @@ -16,4 +16,4 @@ jobs: permissions: contents: read with: - ref: ${{ github.event.pull_request.head.sha || github.sha }} + ref: ${{ github.event.pull_request.head.sha || github.sha }} \ No newline at end of file diff --git a/.github/workflows/pypi-publish-on-release.yml b/.github/workflows/pypi-publish-on-release.yml index a702511..07717c2 100644 --- a/.github/workflows/pypi-publish-on-release.yml +++ b/.github/workflows/pypi-publish-on-release.yml @@ -22,37 +22,37 @@ jobs: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 - with: - persist-credentials: false + - uses: actions/checkout@v4 + with: + persist-credentials: false - - name: Set up Python - uses: actions/setup-python@v5 - with: - python-version: '3.10' + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.10' - - name: Install dependencies - run: | - python -m pip install --upgrade pip - pip install hatch twine - - name: Validate version - run: | - version=$(hatch version) - if [[ $version =~ ^[0-9]+\.[0-9]+\.[0-9]+$ ]]; then - echo "Valid version format" - exit 0 - else - echo "Invalid version format" - exit 1 - fi - - name: Build - run: | - hatch build - - name: Store the distribution packages - uses: actions/upload-artifact@v4 - with: - name: python-package-distributions - path: dist/ + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install hatch twine + - name: Validate version + run: | + version=$(hatch version) + if [[ $version =~ ^[0-9]+\.[0-9]+\.[0-9]+$ ]]; then + echo "Valid version format" + exit 0 + else + echo "Invalid version format" + exit 1 + fi + - name: Build + run: | + hatch build + - name: Store the distribution packages + uses: actions/upload-artifact@v4 + with: + name: python-package-distributions + path: dist/ deploy: name: Upload release to PyPI @@ -70,10 +70,10 @@ jobs: id-token: write steps: - - name: Download all the dists - uses: actions/download-artifact@v4 - with: - name: python-package-distributions - path: dist/ - - name: Publish distribution 📦 to PyPI - uses: pypa/gh-action-pypi-publish@release/v1 \ No newline at end of file + - name: Download all the dists + uses: actions/download-artifact@v4 + with: + name: python-package-distributions + path: dist/ + - name: Publish distribution 📦 to PyPI + uses: pypa/gh-action-pypi-publish@release/v1 \ No newline at end of file diff --git a/.github/workflows/test-lint.yml b/.github/workflows/test-lint.yml index 178b1bb..009eb6f 100644 --- a/.github/workflows/test-lint.yml +++ b/.github/workflows/test-lint.yml @@ -1,4 +1,4 @@ -name: Lint +name: Test and Lint on: workflow_call: @@ -8,6 +8,65 @@ on: type: string jobs: + unit-test: + name: Unit Tests - Python ${{ matrix.python-version }} - ${{ matrix.os-name }} + permissions: + contents: read + strategy: + matrix: + include: + # Linux + - os: ubuntu-latest + os-name: 'linux' + python-version: "3.10" + - os: ubuntu-latest + os-name: 'linux' + python-version: "3.11" + - os: ubuntu-latest + os-name: 'linux' + python-version: "3.12" + - os: ubuntu-latest + os-name: 'linux' + python-version: "3.13" + # Windows + - os: windows-latest + os-name: 'windows' + python-version: "3.10" + - os: windows-latest + os-name: 'windows' + python-version: "3.11" + - os: windows-latest + os-name: 'windows' + python-version: "3.12" + - os: windows-latest + os-name: 'windows' + python-version: "3.13" + # MacOS - latest only; not enough runners for macOS + - os: macos-latest + os-name: 'macOS' + python-version: "3.13" + fail-fast: true + runs-on: ${{ matrix.os }} + env: + LOG_LEVEL: DEBUG + steps: + - name: Checkout code + uses: actions/checkout@v4 + with: + ref: ${{ inputs.ref }} # Explicitly define which commit to check out + persist-credentials: false # Don't persist credentials for subsequent actions + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - name: Install dependencies + run: | + pip install --no-cache-dir hatch + - name: Run Unit tests + id: tests + run: hatch test tests --cover + continue-on-error: false + lint: name: Lint runs-on: ubuntu-latest @@ -32,5 +91,5 @@ jobs: - name: Run lint id: lint - run: hatch run lint - continue-on-error: false + run: hatch run test-lint + continue-on-error: false \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 6b9c0b6..ca8cd2c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,6 +24,7 @@ classifiers = [ dependencies = [ "mcp>=1.1.3", "pydantic>=2.0.0", + "jinja2>=3.1.0", ] [project.scripts] @@ -51,6 +52,12 @@ dev = [ "hatch>=1.0.0", "pre-commit>=2.20.0", "ruff>=0.4.4", + "pytest>=8.0.0", + "pytest-asyncio>=0.23.0", + "pytest-cov>=4.1.0", + "pytest-xdist>=3.0.0", + "pytest-mock>=3.12.0", + "mypy>=0.981", ] [tool.hatch.build] @@ -60,7 +67,9 @@ packages = ["src/strands_mcp_server"] dependencies = [ "mcp>=1.1.3", "pydantic>=2.0.0", + "jinja2>=3.1.0", "ruff>=0.4.4", + "mypy>=0.981", ] [tool.hatch.envs.hatch-static-analysis.scripts] @@ -77,6 +86,25 @@ lint-fix = [ "ruff check --fix" ] +[tool.hatch.envs.hatch-test] +extra-dependencies = [ + "pytest>=8.0.0", + "pytest-cov>=4.1.0", + "pytest-xdist>=3.0.0", + "pytest-asyncio>=0.23.0", + "pytest-mock>=3.12.0", +] +extra-args = ["-n", "auto", '-vv'] + +[[tool.hatch.envs.hatch-test.matrix]] +python = ["3.10", "3.11", "3.12", "3.13"] + +[tool.hatch.envs.hatch-test.scripts] +run = ["pytest{env:HATCH_TEST_ARGS:} {args}"] +run-cov = ["pytest{env:HATCH_TEST_ARGS:} --cov --cov-config=pyproject.toml {args}"] +cov-combine = [] +cov-report = [] + [tool.hatch.envs.default.scripts] list = [ "echo 'Scripts commands available for default env:'; hatch env show --json | jq --raw-output '.default.scripts | keys[]'" @@ -84,13 +112,22 @@ list = [ format = [ "hatch fmt --formatter", ] +test-format = ["hatch fmt --formatter --check"] lint = [ "hatch fmt --linter" ] +test-lint = ["hatch fmt --linter --check"] +test = ["hatch test --cover --cov-report html --cov-report xml {args}"] +prepare = [ + "hatch run format", + "hatch run lint", + "hatch run test-lint", + "hatch run test" +] [tool.ruff] line-length = 120 -include = ["src/**/*.py"] +include = ["src/**/*.py", "tests/**/*.py"] [tool.ruff.lint] select = [ @@ -100,6 +137,39 @@ select = [ "B", # flake8-bugbear ] +[tool.coverage.run] +branch = true +source = ["src"] + +[tool.coverage.report] +show_missing = true + +[tool.coverage.html] +directory = "build/coverage/html" + +[tool.coverage.xml] +output = "build/coverage/coverage.xml" + +[tool.pytest.ini_options] +testpaths = ["tests"] +pythonpath = ["src"] + +[tool.mypy] +python_version = "3.10" +warn_return_any = true +warn_unused_configs = true +disallow_untyped_defs = true +disallow_incomplete_defs = true +check_untyped_defs = true +disallow_untyped_decorators = true +no_implicit_optional = true +warn_redundant_casts = true +warn_unused_ignores = true +warn_no_return = true +warn_unreachable = true +follow_untyped_imports = true +ignore_missing_imports = false + [tool.commitizen] name = "cz_conventional_commits" tag_format = "v$version" diff --git a/src/strands_mcp_server/prompts.py b/src/strands_mcp_server/prompts.py new file mode 100644 index 0000000..89532af --- /dev/null +++ b/src/strands_mcp_server/prompts.py @@ -0,0 +1,268 @@ +"""Prompt generation system for Strands development.""" + +import re +from pathlib import Path + +import jinja2 + +from .utils import cache + +# Initialize Jinja2 environment +template_dir = Path(__file__).parent / "prompts" +jinja_env = jinja2.Environment( + loader=jinja2.FileSystemLoader(template_dir), trim_blocks=True, lstrip_blocks=True, autoescape=False +) + + +def regex_replace_filter(text: str, pattern: str, replacement: str = "") -> str: + """Custom Jinja2 filter for regex replacement. + + Args: + text: The text to process + pattern: The regex pattern to match + replacement: The replacement string (default: empty string) + + Returns: + Text with pattern replaced + """ + if not text: + return text + return re.sub(pattern, replacement, text, flags=re.DOTALL) + + +# Add custom filters to Jinja environment +jinja_env.filters["regex_replace"] = regex_replace_filter + + +def fetch_content(url: str) -> str: + """Fetch content from documentation URL using cache. + + Args: + url: URL to fetch content from + + Returns: + Content string or empty string if fetch fails + """ + cache.ensure_ready() + page = cache.ensure_page(url) + if page and page.content: + return page.content + return "" + + +# ============================================================================ +# SIMPLE PROMPT GENERATORS - Direct documentation inclusion +# ============================================================================ + + +def generate_tool_prompt(request: str, tool_use_examples: str = "", preferred_libraries: str = "") -> str: + """Generate a design-first tool development prompt with dynamic documentation.""" + + cache.ensure_ready() + + # Fetch documentation + llms_txt_url = "https://strandsagents.com/llms.txt" + llms_txt_content = fetch_content(llms_txt_url) + + python_tools_url = ( + "https://strandsagents.com/latest/documentation/docs/user-guide/concepts/tools/python-tools/index.md" + ) + python_tools_content = fetch_content(python_tools_url) + + # Load and render template + template = jinja_env.get_template("tool_development.jinja2") + + return template.render( + request=request, + tool_use_examples=tool_use_examples, + preferred_libraries=preferred_libraries, + llms_txt_content=llms_txt_content, + python_tools_content=python_tools_content, + ) + + +def generate_session_prompt(request: str, include_examples: bool = True) -> str: + """Generate a session management implementation prompt with documentation.""" + + cache.ensure_ready() + + # Fetch documentation + llms_txt_url = "https://strandsagents.com/llms.txt" + llms_txt_content = fetch_content(llms_txt_url) + + session_management_url = ( + "https://strandsagents.com/latest/documentation/docs/user-guide/concepts/agents/session-management/index.md" + ) + session_management_content = fetch_content(session_management_url) + + session_api_url = "https://strandsagents.com/latest/documentation/docs/api-reference/session/index.md" + session_api_content = fetch_content(session_api_url) + + # Load and render template + template = jinja_env.get_template("session_management.jinja2") + + return template.render( + request=request, + include_examples=include_examples, + llms_txt_content=llms_txt_content, + session_management_content=session_management_content, + session_api_content=session_api_content, + ) + + +def generate_agent_prompt( + use_case: str, + examples: str = "", + agent_guidelines: str = "", + tools_required: str = "", + model_preferences: str = "", + include_examples: bool = True, + verbosity: str = "normal", +) -> str: + """Generate a design-first agent development prompt with dynamic documentation. + + Args: + use_case: Description of what the agent should do + examples: Optional examples of expected agent behavior + agent_guidelines: Optional specific behavioral guidelines + tools_required: Optional list of required tools/capabilities + model_preferences: Optional model provider preferences + include_examples: Whether to include code examples + verbosity: Level of detail (minimal, normal, detailed) + + Returns: + Generated prompt text for agent development + """ + + cache.ensure_ready() + + # Fetch documentation + llms_txt_url = "https://strandsagents.com/llms.txt" + llms_txt_content = fetch_content(llms_txt_url) + + agent_loop_url = ( + "https://strandsagents.com/latest/documentation/docs/user-guide/concepts/agents/agent-loop/index.md" + ) + agent_loop_content = fetch_content(agent_loop_url) + + agent_api_url = "https://strandsagents.com/latest/documentation/docs/api-reference/agent/index.md" + agent_api_content = fetch_content(agent_api_url) + + # Fetch community tools documentation - just pass the raw content + community_tools_url = ( + "https://strandsagents.com/latest/documentation/docs/user-guide/concepts/tools/community-tools-package/index.md" + ) + community_tools_content = fetch_content(community_tools_url) + + # Load and render template + template = jinja_env.get_template("agent_development.jinja2") + + return template.render( + use_case=use_case, + examples=examples, + agent_guidelines=agent_guidelines, + tools_required=tools_required, + model_preferences=model_preferences, + llms_txt_content=llms_txt_content, + agent_loop_content=agent_loop_content, + agent_api_content=agent_api_content, + community_tools_content=community_tools_content, # Pass raw content + include_examples=include_examples, + verbosity=verbosity, + ) + + +def generate_model_prompt( + use_case: str, + model_details: str = "", + api_documentation: str = "", + auth_requirements: str = "", + special_features: str = "", + include_examples: bool = True, +) -> str: + """Generate a design-first custom model provider development prompt. + + Args: + use_case: Description of the model provider's purpose + model_details: Details about the models to support + api_documentation: Reference to API docs or endpoints + auth_requirements: Authentication/authorization details + special_features: Special capabilities needed (streaming, function calling, etc.) + include_examples: Whether to include code examples + + Returns: + Generated prompt text for model provider development + """ + + cache.ensure_ready() + + # Fetch documentation + custom_model_url = "https://strandsagents.com/latest/documentation/docs/user-guide/concepts/model-providers/custom_model_provider/index.md" + custom_model_content = fetch_content(custom_model_url) + + models_api_url = "https://strandsagents.com/latest/documentation/docs/api-reference/models/index.md" + models_api_content = fetch_content(models_api_url) + + # Load and render template + template = jinja_env.get_template("model_development.jinja2") + + return template.render( + use_case=use_case, + model_details=model_details, + api_documentation=api_documentation, + auth_requirements=auth_requirements, + special_features=special_features, + custom_model_content=custom_model_content, + models_api_content=models_api_content, + include_examples=include_examples, + ) + + +def generate_multiagent_prompt( + use_case: str, + pattern: str = "", + agent_roles: str = "", + interaction_requirements: str = "", + scale_requirements: str = "", + include_examples: bool = True, +) -> str: + """Generate a multi-agent systems development prompt. + + Args: + use_case: Description of what the multi-agent system should accomplish + pattern: Preferred pattern (graph/swarm/hybrid) or let the prompt guide selection + agent_roles: Description of the different agent roles and specializations needed + interaction_requirements: How agents should interact and collaborate + scale_requirements: Performance, reliability, and scalability requirements + include_examples: Whether to include code examples + + Returns: + A comprehensive prompt for multi-agent system development + """ + + cache.ensure_ready() + + # Load the template + template = jinja_env.get_template("multiagent_development.jinja2") + + # Fetch multi-agent documentation + graph_url = "https://strandsagents.com/latest/documentation/docs/user-guide/concepts/multi-agent/graph/index.md" + swarm_url = "https://strandsagents.com/latest/documentation/docs/user-guide/concepts/multi-agent/swarm/index.md" + multiagent_api_url = "https://strandsagents.com/latest/documentation/docs/api-reference/multiagent/index.md" + + graph_content = fetch_content(graph_url) + swarm_content = fetch_content(swarm_url) + multiagent_api_content = fetch_content(multiagent_api_url) + + # Simply pass all parameters to template - no string building in Python! + return template.render( + use_case=use_case, + pattern=pattern, + agent_roles=agent_roles, + interaction_requirements=interaction_requirements, + scale_requirements=scale_requirements, + graph_content=graph_content, + swarm_content=swarm_content, + multiagent_api_content=multiagent_api_content, + include_examples=include_examples, + ) diff --git a/src/strands_mcp_server/prompts/agent_development.jinja2 b/src/strands_mcp_server/prompts/agent_development.jinja2 new file mode 100644 index 0000000..50cf71a --- /dev/null +++ b/src/strands_mcp_server/prompts/agent_development.jinja2 @@ -0,0 +1,348 @@ +# Strands Agent Development - Design First Approach + +## What is Strands? + +{{ llms_txt_content | default("Strands is a powerful agent framework that enables AI agents to use tools, interact with systems, and accomplish complex tasks autonomously. Agents in Strands follow the agent loop pattern: receiving input, reasoning, selecting tools, executing actions, and generating responses.", true) }} + +## Instructions for the Agent + +You are helping to develop a Strands AI agent. Follow this structured approach: + +### PHASE 1: DESIGN (Do this first) + +1. **Summarize your understanding** of the agent requirements +2. **Design the agent architecture** including: + - Agent name and description + - System prompt (agent's role, expertise, and behavior) + - Tool selection and justification + - Model provider selection + - Session/state management needs +3. **Present the design** to the user for approval + +### PHASE 2: IMPLEMENTATION (Only after user approves the design) + +Once the user approves the design, implement the full agent code using Strands patterns. + +--- + +## User's Requirements + +### Use Case: +{{ use_case }} + +{% if examples %} +### Examples of Expected Behavior: +{{ examples }} + +{% endif %} +{% if agent_guidelines %} +### Specific Agent Guidelines: +{{ agent_guidelines }} + +{% endif %} +{% if tools_required %} +### Required Tools/Capabilities: +{{ tools_required }} + +{% endif %} +{% if model_preferences %} +### Model Preferences: +{{ model_preferences }} + +{% endif %} +--- + +## Your Task (Design Phase) + +Based on the requirements above, please: + +1. **Summarize Your Understanding** + - What is the primary purpose of this agent? + - What tasks should it be able to accomplish? + - What are the key behaviors and constraints? + - Who is the target user? + +2. **Design the Agent Architecture** + + a) **Agent Configuration** + ```python + from strands import Agent + + agent = Agent( + name="agent_name", # Descriptive name + description="What this agent does", # Clear description + agent_id="unique_id", # Optional unique identifier + model="model_id", # Or model provider object + # Additional configuration... + ) + ``` + + b) **System Prompt Design** + ``` + System Prompt: + """ + [Role & Identity] + [Expertise & Knowledge] + [Behavioral Guidelines] + [Response Style] + [Constraints & Ethics] + """ + ``` + + c) **Tool Selection** + - List selected tools with justification + - Explain why each tool is necessary + - Note any custom tools that need development + + d) **Model Provider Selection** + - Recommended model provider and why + - Specific model ID if applicable + - Temperature and other parameters + +3. **Design Decisions** + - Will this agent need session management for stateful conversations? + - What conversation management strategy (sliding window, summarizing, null)? + - Does it need hooks for custom behavior? + - Any special error handling requirements? + - Security or safety considerations? + +4. **Usage Example** + Show how the agent would be initialized and used: + ```python + # Initialize the agent + agent = Agent(...) + + # Use the agent + response = agent("User query here") + print(response) + ``` + +**IMPORTANT**: Present this design and wait for user approval before implementing the full code. + +--- + +## System Prompt Development Guidelines + +### Effective System Prompts Should Include: + +1. **Role & Identity** + - Who/what the agent is + - Its primary function + - Level of expertise + +2. **Expertise & Knowledge** + - Specific domains of knowledge + - Technical capabilities + - Limitations to acknowledge + +3. **Behavioral Guidelines** + - How to approach problems + - Decision-making criteria + - When to use specific tools + +4. **Response Style** + - Tone (professional, friendly, technical) + - Level of detail + - Formatting preferences + +5. **Constraints & Ethics** + - What the agent should NOT do + - Safety considerations + - Privacy and security rules + +### System Prompt Examples: + +**Technical Assistant:** +``` +You are a senior software engineer specializing in Python and cloud architectures. +You provide detailed, accurate technical guidance with code examples. +Always consider security best practices and explain trade-offs in your recommendations. +Use tools to verify information when needed. Be concise but thorough. +``` + +**Research Analyst:** +``` +You are an expert research analyst with deep knowledge of data analysis and reporting. +You excel at finding, synthesizing, and presenting information from multiple sources. +Always cite your sources and distinguish between facts and interpretations. +Use retrieval tools to gather current information and calculator for any computations. +Present findings in a structured, easy-to-understand format. +``` + +--- + +## Tool Selection Guidelines + +### Core Principles: + +1. **Minimal Tool Set**: Only include tools the agent actually needs +2. **Tool Synergy**: Select tools that work well together +3. **Safety First**: Consider security implications of each tool +4. **Performance**: More tools = slower initialization + +### Available Built-in Tools: + +{% if community_tools_content %} +{{ community_tools_content }} +{% else %} +**Information & Search:** +- `retrieve` - Semantically retrieve data from Amazon Bedrock Knowledge Bases for RAG, memory, and other purposes +- `memory` - Agent memory persistence in Amazon Bedrock Knowledge Bases +- `http_request` - Make API calls, fetch web data, and call local HTTP servers +- `file_read` - Read and parse files + +**Content Creation:** +- `file_write` - Create and modify files +- `editor` - File editing operations like line edits, search, and undo +- `generate_image` - Create AI generated images with Amazon Bedrock +- `nova_reels` - Create AI generated videos with Nova Reels on Amazon Bedrock + +**Computation & Analysis:** +- `calculator` - Perform mathematical operations +- `python_repl` - Run Python code +- `think` - Perform deep thinking by creating parallel branches of agentic reasoning + +**System & Environment:** +- `shell` - Execute shell commands +- `environment` - Manage environment variables +- `current_time` - Get the current date and time +- `cron` - Task scheduling with cron jobs + +**Communication:** +- `slack` - Slack integration with real-time events, API access, and message sending +- `speak` - Generate speech from text using macOS say command or Amazon Polly + +**Agent Capabilities:** +- `use_llm` - Run a new AI event loop with custom prompts +- `swarm` - Coordinate multiple AI agents in a swarm / network of agents +- `workflow` - Orchestrate sequenced workflows +- `agent_graph` - Create and manage graphs of agents +- `graph` - Create and manage multi-agent systems using Strands SDK Graph implementation + +**Utility:** +- `journal` - Create structured tasks and logs for agents to manage and work from +- `stop` - Force stop the agent event loop +- `load_tool` - Dynamically load more tools at runtime +- `sleep` - Pause execution with interrupt support +{% endif %} + +### Tool Selection Patterns: + +**Data Analysis Agent:** +```python +tools = ["file_read", "calculator", "python_repl", "file_write"] +``` + +**Research Agent:** +```python +tools = ["retrieve", "http_request", "memory", "file_write"] +``` + +**DevOps Agent:** +```python +tools = ["shell", "file_read", "editor", "environment", "cron"] +``` + +**Creative Agent:** +```python +tools = ["generate_image", "nova_reels", "file_write", "think"] +``` + +**Note:** Custom tools can be developed if built-in tools don't meet requirements. Use the `load_tool` capability or place tools in `./tools/` directory with `load_tools_from_directory=True`. + +--- + +## Strands Agent Documentation + +{% if agent_loop_content %} +{{ agent_loop_content }} +{% else %} +### Agent Loop Overview + +The agent loop is the core execution model: + +1. **Receives user input** and contextual information +2. **Processes the input** using a language model (LLM) +3. **Decides** whether to use tools to gather information or perform actions +4. **Executes tools** and receives results +5. **Continues reasoning** with the new information +6. **Produces a final response** or iterates again through the loop + +This cycle may repeat multiple times within a single interaction, enabling complex multi-step reasoning and autonomous behavior. +{% endif %} + +--- + +## Strands Agent API Reference + +{% if agent_api_content %} +{% set filtered_content = agent_api_content | regex_replace('````.*?````', '') %} +{{ filtered_content }} +{% else %} +### Basic Agent Configuration + +```python +from strands import Agent +from strands.agent import SlidingWindowConversationManager + +agent = Agent( + # Model configuration + model="claude-sonnet-4", # or model provider object + + # Agent identity + name="My Agent", + description="What this agent does", + agent_id="unique_id", + + # Behavior configuration + system_prompt="You are a helpful assistant", + tools=["tool1", "tool2"], # List of tool names + + # Conversation management + conversation_manager=SlidingWindowConversationManager(window_size=40), + + # Advanced options + load_tools_from_directory=True, # Auto-load from ./tools/ + record_direct_tool_call=True, # Record tool.method() calls + state={"key": "value"}, # Initial state + hooks=[custom_hook], # Event hooks + session_manager=session_manager, # For persistence +) + +# Use the agent +response = agent("Your query here") +print(response) + +# Direct tool access +result = agent.tool.calculator(expression="2+2") + +# Structured output +from pydantic import BaseModel + +class Analysis(BaseModel): + summary: str + score: float + +result = agent.structured_output(Analysis, "Analyze this...") +``` + +### Model Providers + +- **Amazon Bedrock**: Default provider, extensive model selection +- **Anthropic**: Claude models, high-quality reasoning +- **OpenAI**: GPT models, broad capabilities +- **Ollama**: Local models, privacy-focused +- **LiteLLM**: Universal interface to multiple providers +- **Custom**: Implement your own provider + +### Documentation References +- Agent Loop: https://strandsagents.com/latest/documentation/docs/user-guide/concepts/agents/agent-loop/index.md +- Agent API: https://strandsagents.com/latest/documentation/docs/api-reference/agent/index.md +- Conversation Management: https://strandsagents.com/latest/documentation/docs/user-guide/concepts/agents/conversation-management/index.md +- Session Management: https://strandsagents.com/latest/documentation/docs/user-guide/concepts/agents/session-management/index.md +- State Management: https://strandsagents.com/latest/documentation/docs/user-guide/concepts/agents/state/index.md +{% endif %} + +--- + +**Remember**: Focus on DESIGN first. Get approval. Then implement the full solution. \ No newline at end of file diff --git a/src/strands_mcp_server/prompts/base.jinja2 b/src/strands_mcp_server/prompts/base.jinja2 new file mode 100644 index 0000000..fb3a45e --- /dev/null +++ b/src/strands_mcp_server/prompts/base.jinja2 @@ -0,0 +1,4 @@ +{# Base template for Strands development prompts #} +{% block content %} +{# Main content block - override in child templates #} +{% endblock %} \ No newline at end of file diff --git a/src/strands_mcp_server/prompts/model_development.jinja2 b/src/strands_mcp_server/prompts/model_development.jinja2 new file mode 100644 index 0000000..60b246c --- /dev/null +++ b/src/strands_mcp_server/prompts/model_development.jinja2 @@ -0,0 +1,310 @@ +# Strands Custom Model Provider Development - Design First Approach + +## Overview + +Strands Agents SDK provides an extensible interface for implementing custom model providers, allowing organizations to integrate their own LLM services, proprietary models, or specialized inference endpoints while maintaining clean separation of concerns. + +## Instructions for Development + +You are helping to develop a custom Strands model provider. Follow this structured approach: + +### PHASE 1: DESIGN (Do this first) + +1. **Summarize your understanding** of the model provider requirements +2. **Design the model provider architecture** including: + - Provider name and description + - Model identification strategy (model IDs, versioning) + - Authentication mechanism (API keys, tokens, custom auth) + - Configuration parameters (endpoints, timeouts, retry logic) + - Request/response transformation needs + - Error handling strategy + - Special features (streaming, function calling, embeddings) +3. **Present the design** to the user for approval + +### PHASE 2: IMPLEMENTATION (Only after user approves the design) + +Once the user approves the design, implement the full model provider code using Strands patterns. + +--- + +## User's Requirements + +### Model Provider Use Case: +{{ use_case }} + +{% if model_details %} +### Model Details: +{{ model_details }} +{% endif %} + +{% if api_documentation %} +### API Documentation/Reference: +{{ api_documentation }} +{% endif %} + +{% if auth_requirements %} +### Authentication Requirements: +{{ auth_requirements }} +{% endif %} + +{% if special_features %} +### Special Features Required: +{{ special_features }} +{% endif %} + +--- + +## Model Provider Architecture + +### Core Components + +A Strands model provider must implement: + +1. **Model Class**: Inherits from `strands.models.Model` +2. **Request Processing**: Transform Strands messages to provider format +3. **Response Handling**: Convert provider responses to Strands format +4. **Error Management**: Handle API errors gracefully +5. **Configuration**: Accept provider-specific parameters + +### Implementation Patterns + +{# Filter out 4-backtick and 5-backtick blocks #} +{% set filtered_api_step1 = models_api_content | regex_replace('`````.*?`````', '') %} +{% set filtered_api = filtered_api_step1 | regex_replace('````.*?````', '') %} +{{ filtered_api | truncate(15000, True, '...') }} + +--- + +## Custom Model Provider Guide + +{{ custom_model_content | truncate(10000, True, '...') }} + +--- + +## Example Model Provider Structure + +```python +from typing import Any, Dict, List, Optional +from strands.models import Model +from strands.models.messages import ( + ContentBlock, + ConversationMessage, + Message, + StopReason, + ToolCall, + ToolResult, + Usage +) + +class CustomModelProvider(Model): + """Custom model provider for [Provider Name].""" + + def __init__( + self, + model_id: str, + api_key: Optional[str] = None, + endpoint: Optional[str] = None, + timeout: int = 30, + max_retries: int = 3, + **kwargs: Any + ): + """Initialize the custom model provider. + + Args: + model_id: Model identifier + api_key: API key for authentication + endpoint: API endpoint URL + timeout: Request timeout in seconds + max_retries: Maximum number of retry attempts + **kwargs: Additional provider-specific parameters + """ + super().__init__(model_id=model_id, **kwargs) + self.api_key = api_key or os.environ.get("CUSTOM_API_KEY") + self.endpoint = endpoint or "https://api.custom-provider.com" + self.timeout = timeout + self.max_retries = max_retries + + def _convert_messages(self, messages: List[Message]) -> List[Dict]: + """Convert Strands messages to provider format.""" + # Implementation here + pass + + def _parse_response(self, response: Dict) -> ConversationMessage: + """Parse provider response to Strands format.""" + # Implementation here + pass + + def generate( + self, + messages: List[Message], + max_tokens: int = 1000, + temperature: float = 0.7, + **kwargs: Any + ) -> ConversationMessage: + """Generate a response from the model. + + Args: + messages: Input messages + max_tokens: Maximum tokens to generate + temperature: Sampling temperature + **kwargs: Additional generation parameters + + Returns: + Generated conversation message + """ + # Convert messages to provider format + provider_messages = self._convert_messages(messages) + + # Make API request with retry logic + for attempt in range(self.max_retries): + try: + response = self._make_request( + messages=provider_messages, + max_tokens=max_tokens, + temperature=temperature, + **kwargs + ) + break + except Exception as e: + if attempt == self.max_retries - 1: + raise + time.sleep(2 ** attempt) + + # Parse and return response + return self._parse_response(response) +``` + +--- + +## Key Implementation Considerations + +### 1. Message Format Conversion +- Transform between Strands message format and provider-specific format +- Handle text, images, tool calls, and tool results +- Preserve message metadata and context + +### 2. Authentication & Security +- Support multiple authentication methods (API keys, OAuth, custom) +- Secure credential storage and management +- Rate limiting and quota management + +### 3. Error Handling +- Graceful degradation for API failures +- Retry logic with exponential backoff +- Meaningful error messages for debugging + +### 4. Performance Optimization +- Connection pooling for HTTP clients +- Response streaming for long outputs +- Caching for repeated requests + +### 5. Observability +- Logging of requests and responses +- Metrics collection (latency, token usage, errors) +- Debugging helpers for development + +{% if include_examples %} +## Advanced Examples + +### Streaming Support + +```python +def generate_stream( + self, + messages: List[Message], + **kwargs: Any +) -> Iterator[ConversationMessage]: + """Generate streaming responses.""" + provider_messages = self._convert_messages(messages) + + # Stream from provider + for chunk in self._stream_request(provider_messages, **kwargs): + yield self._parse_stream_chunk(chunk) +``` + +### Function Calling + +```python +def _handle_tool_calls(self, response: Dict) -> List[ToolCall]: + """Extract and format tool calls from response.""" + tool_calls = [] + if "functions" in response: + for func in response["functions"]: + tool_calls.append(ToolCall( + id=func["id"], + name=func["name"], + input=func["arguments"] + )) + return tool_calls +``` + +### Custom Retry Logic + +```python +def _make_request_with_retry(self, **kwargs) -> Dict: + """Make API request with custom retry logic.""" + for attempt in range(self.max_retries): + try: + return self._make_request(**kwargs) + except RateLimitError as e: + wait_time = e.retry_after or (2 ** attempt) + time.sleep(wait_time) + except TemporaryError as e: + if attempt == self.max_retries - 1: + raise + time.sleep(2 ** attempt) +``` +{% endif %} + +--- + +## Testing Your Model Provider + +```python +from strands import Agent + +# Test the custom provider +agent = Agent( + model=CustomModelProvider( + model_id="custom-model-v1", + api_key="your-api-key", + endpoint="https://api.custom.com" + ), + system_prompt="You are a helpful assistant." +) + +# Test basic generation +response = agent.generate("Hello, how are you?") +print(response) + +# Test with tools +from strands import tool + +@tool +def get_weather(location: str) -> str: + return f"Weather in {location}: Sunny, 72°F" + +agent_with_tools = Agent( + model=CustomModelProvider(model_id="custom-model-v1"), + tools=[get_weather] +) + +response = agent_with_tools.generate("What's the weather in Paris?") +``` + +--- + +## Best Practices + +1. **Type Safety**: Use type hints throughout your implementation +2. **Documentation**: Provide clear docstrings for all methods +3. **Testing**: Include unit tests and integration tests +4. **Examples**: Provide usage examples in your documentation +5. **Error Messages**: Make errors actionable with clear messages +6. **Logging**: Use appropriate log levels for debugging +7. **Configuration**: Support both code and environment variable configuration +8. **Backwards Compatibility**: Version your provider appropriately + +--- + +Now, based on your requirements, I'll help you design and implement a custom model provider that integrates seamlessly with the Strands Agents SDK. \ No newline at end of file diff --git a/src/strands_mcp_server/prompts/multiagent_development.jinja2 b/src/strands_mcp_server/prompts/multiagent_development.jinja2 new file mode 100644 index 0000000..65d1924 --- /dev/null +++ b/src/strands_mcp_server/prompts/multiagent_development.jinja2 @@ -0,0 +1,514 @@ +{% extends "base.jinja2" %} + +{% block content %} +# Strands Multi-Agent Systems Development - Design First Approach + +## Overview + +Strands Agents SDK provides two powerful patterns for building multi-agent systems: **Graph** for structured workflows and **Swarm** for autonomous collaboration. This guide helps you design and implement the right multi-agent architecture for your use case. + +## User's Requirements + +### Use Case: +{{ use_case }} + +{% if pattern %} +## Preferred Pattern: {{ pattern | upper }} + +{% if pattern | lower == "graph" %} +The user has indicated a preference for the Graph pattern. Focus on structured workflows, explicit routing, and dependency management. +{% elif pattern | lower == "swarm" %} +The user has indicated a preference for the Swarm pattern. Focus on autonomous collaboration, dynamic handoffs, and emergent behavior. +{% elif pattern | lower == "hybrid" %} +The user wants a hybrid approach. Consider using Graph for overall structure with Swarm nodes for complex sub-tasks that benefit from autonomous collaboration. +{% endif %} +{% endif %} + +{% if agent_roles %} +## Agent Roles and Specializations + +{{ agent_roles }} + +Design agents with focused expertise areas. Each agent should have: +- Clear system prompt defining its role +- Appropriate tools for its responsibilities +- Optimal model selection for its tasks +{% endif %} + +{% if interaction_requirements %} +## Interaction Requirements + +{{ interaction_requirements }} + +Consider these interaction patterns: +- Message format and structure +- Handoff conditions and triggers +- Shared context and memory +- Coordination mechanisms +{% endif %} + +{% if scale_requirements %} +## Scale and Performance Requirements + +{{ scale_requirements }} + +Address these scalability concerns: +- Parallel execution strategies +- Resource optimization +- Error handling and recovery +- Monitoring and observability +{% endif %} + +--- + +## Why Multi-Agent Systems? + +Multi-agent systems solve complex problems that are difficult or impossible for single agents: + +### **Cognitive Load Distribution** +- Break complex tasks into specialized sub-tasks +- Each agent maintains focused expertise +- Reduced context switching and improved accuracy + +### **Parallel Processing** +- Execute independent tasks simultaneously +- Reduce total completion time +- Scale horizontally with task complexity + +### **Specialized Expertise** +- Agents with different system prompts and tools +- Domain-specific knowledge isolation +- Optimized model selection per agent + +### **Fault Tolerance** +- Isolated failure domains +- Graceful degradation +- Retry mechanisms at agent level + +### **Emergent Intelligence** +- Collective problem-solving beyond individual capabilities +- Dynamic adaptation to complex scenarios +- Creative solutions through agent interaction + +## Instructions for Development + +You are helping to develop a multi-agent system using Strands. Follow this structured approach: + +### PHASE 1: DESIGN (Do this first) + +1. **Understand the Requirements** + - What is the main goal of the multi-agent system? + - What are the sub-tasks or specialized roles needed? + - What are the interaction patterns between agents? + - What are the performance and reliability requirements? + +2. **Choose the Right Pattern** + + **Use GRAPH when you need:** + - Structured, predictable workflows + - Explicit control over agent interactions + - Complex routing logic and conditions + - Parallel execution with dependencies + - Visual representation of the system + + **Use SWARM when you need:** + - Autonomous agent collaboration + - Dynamic task delegation + - Self-organizing behavior + - Flexible problem-solving + - Minimal orchestration overhead + +3. **Design the Architecture** + - Define agent roles and responsibilities + - Map out communication patterns + - Identify shared resources and state + - Plan error handling and recovery + - Consider scalability requirements + +4. **Present Your Design** + - System architecture diagram (conceptual) + - Agent specifications + - Interaction patterns + - Expected behavior and outcomes + +### PHASE 2: IMPLEMENTATION + +Only proceed to implementation after design approval. Provide complete, working code with: +- All necessary imports +- Comprehensive error handling +- Type hints and documentation +- Example usage +- Testing considerations +## Graph vs Swarm: Choosing the Right Pattern + +### Graph Pattern +**Best for:** Structured workflows, ETL pipelines, approval chains, complex orchestration + +**Characteristics:** +- Nodes represent agents with specific roles +- Edges define message flow and dependencies +- Supports conditional routing and parallel execution +- Centralized orchestration and monitoring + +**Example Use Cases:** +- Document processing pipelines +- Multi-stage analysis workflows +- Approval and review processes +- Data transformation pipelines + +### Swarm Pattern +**Best for:** Creative problem-solving, research tasks, collaborative analysis, emergent solutions + +**Characteristics:** +- Agents autonomously decide when to collaborate +- Dynamic handoffs based on expertise +- Self-organizing behavior +- Minimal central control + +**Example Use Cases:** +- Complex research projects +- Creative content generation +- Multi-perspective analysis +- Exploratory problem-solving + +{% if graph_content %} +## Graph Pattern - Deep Dive + +{# Filter out 4-backtick and 5-backtick blocks from graph content #} +{% set filtered_graph_step1 = graph_content | regex_replace('`````.*?`````', '') %} +{% set filtered_graph = filtered_graph_step1 | regex_replace('````.*?````', '') %} +{{ filtered_graph }} +{% endif %} + +{% if swarm_content %} +## Swarm Pattern - Deep Dive + +{# Filter out 4-backtick and 5-backtick blocks from swarm content #} +{% set filtered_swarm_step1 = swarm_content | regex_replace('`````.*?`````', '') %} +{% set filtered_swarm = filtered_swarm_step1 | regex_replace('````.*?````', '') %} +{{ filtered_swarm }} +{% endif %} + +{% if multiagent_api_content %} +## Multi-Agent API Reference + +{# Filter out 4-backtick and 5-backtick blocks from API content #} +{% set filtered_api_step1 = multiagent_api_content | regex_replace('`````.*?`````', '') %} +{% set filtered_api = filtered_api_step1 | regex_replace('````.*?````', '') %} +{{ filtered_api }} +{% endif %} + +## Implementation Examples + +### Graph Pattern Example: Research Pipeline + +```python +from strands.agent import Agent +from strands.multiagent.graph import Graph, Node, Edge, Message + +# Define specialized agents +researcher = Agent( + system_prompt="You are a research specialist. Find and summarize relevant information.", + tools=["retrieve", "http_request"] +) + +analyst = Agent( + system_prompt="You are a data analyst. Analyze findings and identify patterns.", + tools=["calculator", "file_write"] +) + +writer = Agent( + system_prompt="You are a technical writer. Create clear, comprehensive reports.", + tools=["file_write", "generate_image"] +) + +# Build the graph +graph = Graph(name="research_pipeline") + +# Add nodes +research_node = Node(name="research", agent=researcher) +analysis_node = Node(name="analysis", agent=analyst) +writing_node = Node(name="writing", agent=writer) + +graph.add_node(research_node) +graph.add_node(analysis_node) +graph.add_node(writing_node) + +# Define edges (workflow) +graph.add_edge(Edge(source="research", target="analysis")) +graph.add_edge(Edge(source="analysis", target="writing")) + +# Execute the pipeline +initial_message = Message( + content="Research renewable energy trends for 2024", + sender="user", + target="research" +) + +result = await graph.execute(initial_message) +``` + +### Swarm Pattern Example: Creative Problem Solving + +```python +from strands.agent import Agent +from strands.multiagent.swarm import Swarm, SwarmConfig + +# Create specialized agents +swarm = Swarm( + name="creative_team", + config=SwarmConfig( + max_handoffs=20, + enable_monitoring=True + ) +) + +# Add agents with different expertise +swarm.add_agent( + name="strategist", + agent=Agent( + system_prompt="You are a strategic thinker. Focus on high-level planning and goals.", + tools=["think", "file_write"] + ) +) + +swarm.add_agent( + name="creative", + agent=Agent( + system_prompt="You are a creative designer. Generate innovative ideas and concepts.", + tools=["generate_image", "file_write"], + model_settings={"temperature": 0.9} + ) +) + +swarm.add_agent( + name="critic", + agent=Agent( + system_prompt="You are a constructive critic. Evaluate ideas and suggest improvements.", + tools=["think", "file_write"] + ) +) + +# Execute with autonomous collaboration +result = await swarm.execute( + "Design an innovative solution for urban transportation" +) +``` + +## Advanced Patterns + +### Hybrid Approach: Graph with Swarm Nodes + +```python +# Use Graph for overall structure, Swarm for complex sub-tasks +graph = Graph(name="hybrid_system") + +# Research phase uses a swarm for comprehensive exploration +research_swarm = Swarm(name="research_team") +research_swarm.add_agent("web_researcher", web_agent) +research_swarm.add_agent("academic_researcher", academic_agent) +research_swarm.add_agent("industry_researcher", industry_agent) + +# Add swarm as a node in the graph +graph.add_node(Node(name="research", agent=research_swarm)) +graph.add_node(Node(name="synthesis", agent=synthesis_agent)) + +# Swarm handles complex research, graph handles workflow +``` + +### Dynamic Agent Creation + +```python +from strands.multiagent.factory import AgentFactory + +# Create agents dynamically based on task requirements +factory = AgentFactory() + +for specialty in task_requirements: + agent = factory.create_agent( + name=f"{specialty}_expert", + system_prompt=f"You are an expert in {specialty}.", + tools=get_tools_for_specialty(specialty) + ) + swarm.add_agent(agent.name, agent) +``` + +## Best Practices + +### 1. **Agent Design** +- Keep agent roles focused and well-defined +- Use clear, specific system prompts +- Assign only necessary tools to each agent +- Consider model selection per agent role + +### 2. **Communication** +- Design clear message formats +- Include context in handoffs +- Implement result validation +- Log all agent interactions + +### 3. **Error Handling** +- Implement timeouts at agent and system level +- Design fallback strategies +- Monitor for circular dependencies +- Handle partial failures gracefully + +### 4. **Performance** +- Use parallel execution where possible +- Implement caching for expensive operations +- Monitor token usage across agents +- Profile and optimize bottlenecks + +### 5. **Testing** +- Unit test individual agents +- Integration test agent interactions +- Test failure scenarios +- Validate emergent behaviors + +## Common Pitfalls and Solutions + +### Pitfall: Circular Dependencies +**Solution:** Use directed graphs, implement cycle detection, set maximum handoff limits + +### Pitfall: Context Loss +**Solution:** Maintain shared memory, include context in messages, use structured data formats + +### Pitfall: Runaway Costs +**Solution:** Set token budgets, use cheaper models for simple tasks, implement cost monitoring + +### Pitfall: Deadlocks +**Solution:** Implement timeouts, use async execution, design clear termination conditions + +## Monitoring and Observability + +```python +from strands.multiagent.monitoring import Monitor + +monitor = Monitor() + +# Track agent interactions +monitor.on_message(lambda msg: print(f"[{msg.sender} -> {msg.target}]: {msg.content[:50]}...")) + +# Track performance metrics +monitor.on_agent_complete(lambda agent, duration, tokens: + print(f"{agent.name}: {duration:.2f}s, {tokens} tokens") +) + +# Attach to your multi-agent system +graph.attach_monitor(monitor) +# or +swarm.attach_monitor(monitor) +``` + +## Production Considerations + +1. **Scalability** + - Horizontal scaling with agent pools + - Queue-based message passing + - Distributed execution support + +2. **Reliability** + - Health checks for agents + - Circuit breakers for failing agents + - Graceful degradation strategies + +3. **Security** + - Agent-level access controls + - Secure message passing + - Audit logging + +4. **Cost Management** + - Token usage tracking + - Model selection optimization + - Caching strategies + +Remember: Start simple, validate the approach, then scale complexity as needed. + +## Complete Implementation Example: Document Analysis System + +```python +from strands.agent import Agent +from strands.multiagent.graph import Graph, Node, Edge, ConditionalEdge +from strands.multiagent.swarm import Swarm +from typing import Dict, Any + +class DocumentAnalysisSystem: + """Multi-agent system for comprehensive document analysis.""" + + def __init__(self): + self.graph = Graph(name="document_analysis") + self._setup_agents() + self._setup_workflow() + + def _setup_agents(self): + """Initialize specialized agents.""" + + # Document processor agent + self.processor = Agent( + system_prompt="""You are a document processing specialist. + Extract and structure key information from documents.""", + tools=["file_read", "file_write"] + ) + + # Analysis swarm for multi-perspective analysis + self.analysis_swarm = Swarm(name="analysis_team") + + self.analysis_swarm.add_agent( + name="financial_analyst", + agent=Agent( + system_prompt="You are a financial analyst. Focus on financial implications and metrics.", + tools=["calculator", "retrieve"] + ) + ) + + self.analysis_swarm.add_agent( + name="risk_analyst", + agent=Agent( + system_prompt="You are a risk analyst. Identify potential risks and mitigation strategies.", + tools=["think", "retrieve"] + ) + ) + + self.analysis_swarm.add_agent( + name="strategy_analyst", + agent=Agent( + system_prompt="You are a strategy analyst. Focus on strategic implications and recommendations.", + tools=["think", "retrieve"] + ) + ) + + # Report generator + self.reporter = Agent( + system_prompt="""You are a report generator. + Create comprehensive reports from analysis results.""", + tools=["file_write", "generate_image"] + ) + + def _setup_workflow(self): + """Configure the processing workflow.""" + + # Add nodes + self.graph.add_node(Node(name="process", agent=self.processor)) + self.graph.add_node(Node(name="analyze", agent=self.analysis_swarm)) + self.graph.add_node(Node(name="report", agent=self.reporter)) + + # Add edges + self.graph.add_edge(Edge(source="process", target="analyze")) + self.graph.add_edge(Edge(source="analyze", target="report")) + + async def analyze_document(self, document_path: str) -> Dict[str, Any]: + """Analyze a document using the multi-agent system.""" + + initial_message = { + "content": f"Analyze the document at {document_path}", + "metadata": {"path": document_path} + } + + result = await self.graph.execute(initial_message) + return result + +# Usage +system = DocumentAnalysisSystem() +result = await system.analyze_document("/path/to/document.pdf") +``` +{% endblock %} \ No newline at end of file diff --git a/src/strands_mcp_server/prompts/session_management.jinja2 b/src/strands_mcp_server/prompts/session_management.jinja2 new file mode 100644 index 0000000..6602320 --- /dev/null +++ b/src/strands_mcp_server/prompts/session_management.jinja2 @@ -0,0 +1,225 @@ +# Strands Session Management - Design First Approach + +## What is Strands? + +Strands is a powerful agent framework that enables AI agents to use tools, interact with systems, and accomplish complex tasks autonomously. Agents in Strands follow the agent loop pattern: receiving input, reasoning, selecting tools, executing actions, and generating responses. + +{{ llms_txt_content }} + +## Instructions for the Agent + +You are helping to implement session management for a Strands application. Follow this structured approach: + +### PHASE 1: DESIGN (Do this first) + +1. **Summarize your understanding** of the session requirements +2. **Design the session architecture** including: + - Session type (basic, persistent, custom) + - Storage mechanism (memory, file, database) + - Session lifecycle (creation, updates, cleanup) + - State management strategy +3. **Present the design** to the user for approval + +### PHASE 2: IMPLEMENTATION (Only after user approves the design) + +Once the user approves the design, implement the full session management code using Strands patterns. + +--- + +## User's Requirements + +### Session Management Request: +{{ request }} + +{% if include_examples %} +### Include Examples: +Yes - Show practical usage examples +{% endif %} + +--- + +## Your Task (Design Phase) + +Based on the requirements above, please: + +1. **Summarize Your Understanding** + - What type of session management is needed? + - What data needs to be stored in the session? + - How long should sessions persist? + - What are the key operations required? + +2. **Design the Session Architecture** + + Create a clear design showing: + + ```python + from strands.session import Session, FileSessionStore + + # Session configuration + session_config = { + "store_type": "file", # or "memory", "custom" + "ttl": 3600, # Time to live in seconds + "max_messages": 100, # Maximum conversation history + # Additional configuration... + } + + # Session implementation approach + class MySessionManager: + """Describe the session management approach.""" + pass + ``` + +3. **Design Decisions** + - What storage backend will be used and why? + - How will session cleanup be handled? + - What about concurrent access to sessions? + - Security considerations? + +4. **Usage Example** + Show how the session will be used: + ```python + # Create or retrieve session + session = get_or_create_session(session_id) + + # Use with agent + agent = Agent(session=session, ...) + response = agent.run("User query") + ``` + +**IMPORTANT**: Present this design and wait for user approval before implementing the full code. + +--- + +## Strands Session Management Documentation + +{% if session_management_content %} +{% set filtered_content = session_management_content | regex_replace('````.*?````', '') %} +{{ filtered_content }} +{% else %} +### Basic Session Patterns + +Sessions in Strands provide stateful conversation management: + +```python +from strands import Agent +from strands.session import Session, FileSessionStore + +# Create a session store +store = FileSessionStore(base_path="./sessions") + +# Create or retrieve a session +session = Session( + session_id="user-123", + store=store +) + +# Use session with agent +agent = Agent( + session=session, + system_prompt="You are a helpful assistant", + tools=["calculator", "file_read"] +) + +# The agent now maintains conversation history +response = agent.run("What's 2+2?") +response = agent.run("Multiply that by 10") # Remembers previous context +``` + +### Session Types + +1. **Memory Session** - Temporary, in-memory storage +2. **File Session** - Persistent file-based storage +3. **Custom Session** - Implement your own storage backend + +### Key Concepts + +- **Session ID**: Unique identifier for each session +- **Message History**: Stored conversation turns +- **Session Store**: Backend storage mechanism +- **TTL**: Time-to-live for session expiration +- **Cleanup**: Automatic or manual session cleanup +{% endif %} + +--- + +## Strands Session API Reference + +{% if session_api_content %} +{% set filtered_content = session_api_content | regex_replace('````.*?````', '') %} +{{ filtered_content }} +{% else %} +### Session Class + +```python +class Session: + """Manages conversation state and history.""" + + def __init__( + self, + session_id: str, + store: Optional[SessionStore] = None, + max_messages: int = 100, + ttl: Optional[int] = None + ): + """Initialize a session.""" + pass + + def add_message(self, role: str, content: str) -> None: + """Add a message to the session history.""" + pass + + def get_messages(self) -> List[Dict]: + """Retrieve all messages in the session.""" + pass + + def clear(self) -> None: + """Clear the session history.""" + pass + + def save(self) -> None: + """Persist the session to storage.""" + pass +``` + +### SessionStore Interface + +```python +class SessionStore(ABC): + """Abstract base class for session storage.""" + + @abstractmethod + def load(self, session_id: str) -> Optional[Dict]: + """Load session data from storage.""" + pass + + @abstractmethod + def save(self, session_id: str, data: Dict) -> None: + """Save session data to storage.""" + pass + + @abstractmethod + def delete(self, session_id: str) -> None: + """Delete a session from storage.""" + pass + + @abstractmethod + def list_sessions(self) -> List[str]: + """List all available session IDs.""" + pass +``` + +### Built-in Stores + +- **MemorySessionStore**: In-memory storage (non-persistent) +- **FileSessionStore**: JSON file-based storage +- **SQLiteSessionStore**: SQLite database storage (requires strands[sqlite]) + +### Documentation References +- Session Management Guide: https://strandsagents.com/latest/documentation/docs/user-guide/concepts/agents/session-management/index.md +- Session API Reference: https://strandsagents.com/latest/documentation/docs/api-reference/session/index.md +- Multi-turn Conversations: https://strandsagents.com/latest/documentation/docs/user-guide/examples/multi-turn/index.md +{% endif %} + +--- + +**Remember**: Focus on DESIGN first. Get approval. Then implement the full solution. \ No newline at end of file diff --git a/src/strands_mcp_server/prompts/tool_development.jinja2 b/src/strands_mcp_server/prompts/tool_development.jinja2 new file mode 100644 index 0000000..2d6a038 --- /dev/null +++ b/src/strands_mcp_server/prompts/tool_development.jinja2 @@ -0,0 +1,198 @@ +# Strands Tool Development - Design First Approach + +## What is Strands? + +{{ llms_txt_content | default("Strands is a powerful agent framework that enables AI agents to use tools, interact with systems, and accomplish complex tasks autonomously.", true) }} + +## Instructions for Tool Development + +You are helping to develop a Strands tool. Follow this structured approach: + +### PHASE 1: DESIGN (Do this first) + +1. **Summarize your understanding** of the tool requirements +2. **Design the tool interface** including: + - Tool name and description + - Input parameters with types and descriptions + - Return value specification + - Error handling approach +3. **Present the design** to the user for approval + +### PHASE 2: IMPLEMENTATION (Only after user approves the design) + +Once the user approves the design, implement the full tool code using the @tool decorator pattern. + +--- + +## User's Requirements + +### Tool Request: +{{ request }} + +{% if tool_use_examples %} +### Usage Examples: +{{ tool_use_examples }} +{% endif %} + +{% if preferred_libraries %} +### Preferred Libraries/APIs: +{{ preferred_libraries }} +{% endif %} + +--- + +## Your Task (Design Phase) + +Based on the requirements above, please: + +1. **Summarize Your Understanding** + - What is the primary purpose of this tool? + - What inputs does it need? + - What outputs should it produce? + - What are the key operations it performs? + +2. **Design the Tool Interface** + + ```python + from strands import tool + + @tool + def tool_name( + param1: str, + param2: int = 42, + # Additional parameters... + ) -> dict: + """ + Tool description - explain what it does. + + Args: + param1: Description of first parameter + param2: Description of second parameter (default: 42) + + Returns: + Dictionary with results in Strands format + """ + # Implementation outline here + pass + ``` + +3. **Design Decisions** + - What libraries or APIs will be used? + - How will errors be handled? + - What validation is needed for inputs? + - Are there any performance considerations? + - Security or safety considerations? + +4. **Usage Example** + Show how the tool would be used: + ```python + # Example usage + result = agent.tool.tool_name( + param1="value", + param2=100 + ) + print(result) + ``` + +**IMPORTANT**: Present this design and wait for user approval before implementing the full code. + +--- + +## Strands Tool Pattern + +### Standard Tool Response Format + +Tools should return a dictionary with status and content: + +```python +@tool +def my_tool(param: str) -> dict: + """Tool description.""" + try: + # Tool implementation + result = do_something(param) + return { + "status": "success", + "content": [{"text": f"Result: {result}"}] + } + except Exception as e: + return { + "status": "error", + "content": [{"text": f"Error: {str(e)}"}] + } +``` + +### Key Principles + +1. **Clear Documentation**: Every parameter and return value must be documented +2. **Type Hints**: Use proper type annotations for all parameters +3. **Error Handling**: Always handle exceptions gracefully +4. **Consistent Format**: Return standardized response dictionaries +5. **Validation**: Validate inputs before processing + +--- + +## Python Tools Documentation + +{% if python_tools_content %} +{{ python_tools_content }} +{% else %} +### Creating Python Tools + +Python tools in Strands use the @tool decorator for simple, type-safe tool creation: + +```python +from strands import tool +from typing import Optional + +@tool +def calculate_compound_interest( + principal: float, + rate: float, + time: int, + compound_per_year: int = 12 +) -> dict: + """ + Calculate compound interest. + + Args: + principal: Initial amount + rate: Annual interest rate (as decimal, e.g., 0.05 for 5%) + time: Time period in years + compound_per_year: Compounding frequency (default: 12 for monthly) + + Returns: + Dictionary with calculation results + """ + try: + amount = principal * (1 + rate/compound_per_year) ** (compound_per_year * time) + interest = amount - principal + + return { + "status": "success", + "content": [{ + "text": f"Final amount: ${amount:.2f}\nInterest earned: ${interest:.2f}" + }] + } + except Exception as e: + return { + "status": "error", + "content": [{"text": f"Calculation error: {str(e)}"}] + } +``` + +### Tool Best Practices + +1. **Single Responsibility**: Each tool should do one thing well +2. **Clear Naming**: Use descriptive, action-oriented names +3. **Comprehensive Docs**: Include examples in docstrings +4. **Input Validation**: Check for valid inputs before processing +5. **Graceful Failures**: Return informative error messages +6. **Testing**: Write unit tests for your tools +7. **Performance**: Consider caching for expensive operations + +{% endif %} + +--- + +**Remember**: Focus on DESIGN first. Get approval. Then implement the full solution. \ No newline at end of file diff --git a/src/strands_mcp_server/server.py b/src/strands_mcp_server/server.py index ff3043b..d765cab 100644 --- a/src/strands_mcp_server/server.py +++ b/src/strands_mcp_server/server.py @@ -2,6 +2,13 @@ from mcp.server.fastmcp import FastMCP +from .prompts import ( + generate_agent_prompt, + generate_model_prompt, + generate_multiagent_prompt, + generate_session_prompt, + generate_tool_prompt, +) from .utils import cache, text_processor APP_NAME = "strands-agents-mcp-server" @@ -73,6 +80,187 @@ def search_docs(query: str, k: int = 5) -> List[Dict[str, Any]]: return return_docs +@mcp.prompt() +def strands_tool_development(request: str, tool_use_examples: str = "", preferred_libraries: str = "") -> str: + """Generate a design-first prompt for developing Strands tools. + + This prompt guides the development process through: + 1. Understanding the user's requirements + 2. Designing the tool interface first + 3. Getting user approval on the design + 4. Implementing the full tool code + + The workflow ensures clear communication and agreement before implementation. + + Args: + request: Description of the tool functionality needed + tool_use_examples: Examples of how the tool will be used (optional) + preferred_libraries: Specific libraries or APIs to use (optional) + + Returns: + A structured prompt for design-first tool development + """ + return generate_tool_prompt(request, tool_use_examples, preferred_libraries) + + +@mcp.prompt() +def strands_agent_development( + use_case: str, examples: str = "", agent_guidelines: str = "", tools_required: str = "", model_preferences: str = "" +) -> str: + """Generate a design-first prompt for developing Strands agents with comprehensive guidelines. + + This prompt guides agent development through: + 1. Understanding the requirements + 2. Designing the agent architecture first (system prompt, tools, model) + 3. Getting user approval on the design + 4. Implementing the full agent code + + The workflow ensures proper agent design with: + - Effective system prompt development + - Strategic tool selection + - Appropriate model provider choice + - Session and state management planning + + Args: + use_case: Description of what the agent should do (e.g., "research assistant for academic papers") + examples: Optional examples of expected agent behavior or interactions + agent_guidelines: Optional specific behavioral guidelines or constraints for the agent + tools_required: Optional list of specific tools or capabilities the agent must have + model_preferences: Optional preferences for model provider or specific models + + Returns: + A structured prompt for design-first agent development + """ + return generate_agent_prompt( + use_case=use_case, + examples=examples, + agent_guidelines=agent_guidelines, + tools_required=tools_required, + model_preferences=model_preferences, + include_examples=bool(examples), + verbosity="normal", + ) + + +@mcp.prompt() +def strands_session_management(request: str, include_examples: bool = True) -> str: + """Generate a design-first prompt for implementing Strands session management. + + This prompt guides the implementation of session management through: + 1. Understanding the session requirements + 2. Designing the session architecture first + 3. Getting user approval on the design + 4. Implementing the full session management code + + Session management enables: + - Stateful conversations with memory + - Multi-turn interactions with context + - Persistent conversation history + - Session lifecycle management + - Custom storage backends + + Args: + request: Description of the session management needs + include_examples: Include usage examples (default: True) + + Returns: + A structured prompt for design-first session implementation + """ + return generate_session_prompt(request, include_examples) + + +@mcp.prompt() +def strands_model_development( + use_case: str, + model_details: str = "", + api_documentation: str = "", + auth_requirements: str = "", + special_features: str = "", +) -> str: + """Generate a design-first prompt for developing custom Strands model providers. + + This prompt guides the development of custom model providers through: + 1. Understanding the model integration requirements + 2. Designing the provider architecture first (auth, API format, features) + 3. Getting user approval on the design + 4. Implementing the full model provider code + + Custom model providers enable: + - Integration with proprietary LLM services + - Custom inference endpoints + - Specialized model configurations + - Organization-specific authentication + - Advanced features (streaming, function calling, embeddings) + + Args: + use_case: Description of the model provider's purpose (e.g., "integrate our company's custom LLM API") + model_details: Optional details about the models to support (versions, capabilities, endpoints) + api_documentation: Optional reference to API documentation or endpoint specifications + auth_requirements: Optional authentication/authorization requirements (API keys, OAuth, custom) + special_features: Optional special capabilities needed (streaming, function calling, embeddings) + + Returns: + A structured prompt for design-first model provider development + """ + return generate_model_prompt( + use_case=use_case, + model_details=model_details, + api_documentation=api_documentation, + auth_requirements=auth_requirements, + special_features=special_features, + include_examples=True, + ) + + +@mcp.prompt() +def strands_multiagent_development( + use_case: str, + pattern: str = "", + agent_roles: str = "", + interaction_requirements: str = "", + scale_requirements: str = "", +) -> str: + """Generate a design-first prompt for developing multi-agent systems with Strands. + + This prompt guides the development of multi-agent systems through: + 1. Understanding why multi-agent is needed for the use case + 2. Choosing between Graph (structured) vs Swarm (autonomous) patterns + 3. Designing the agent architecture and interactions first + 4. Getting user approval on the design + 5. Implementing the full multi-agent system + + Multi-agent systems enable: + - **Cognitive Load Distribution**: Break complex tasks into specialized sub-tasks + - **Parallel Processing**: Execute independent tasks simultaneously + - **Specialized Expertise**: Agents with different prompts, tools, and models + - **Fault Tolerance**: Isolated failure domains with graceful degradation + - **Emergent Intelligence**: Collective problem-solving beyond individual capabilities + + Pattern Selection Guide: + - **Graph**: Use for structured workflows, ETL pipelines, approval chains + - **Swarm**: Use for creative problem-solving, research, collaborative analysis + - **Hybrid**: Combine Graph structure with Swarm nodes for complex sub-tasks + + Args: + use_case: Description of what the multi-agent system should accomplish + pattern: Preferred pattern - "graph", "swarm", or "hybrid" (optional, will guide selection if not specified) + agent_roles: Description of the different agent roles and specializations needed (optional) + interaction_requirements: How agents should interact and collaborate (optional) + scale_requirements: Performance, reliability, and scalability requirements (optional) + + Returns: + A structured prompt for design-first multi-agent system development + """ + return generate_multiagent_prompt( + use_case=use_case, + pattern=pattern, + agent_roles=agent_roles, + interaction_requirements=interaction_requirements, + scale_requirements=scale_requirements, + include_examples=True, + ) + + @mcp.tool() def fetch_doc(uri: str = "") -> Dict[str, Any]: """Fetch full document content by URL. diff --git a/tests/README.md b/tests/README.md new file mode 100644 index 0000000..4fa862f --- /dev/null +++ b/tests/README.md @@ -0,0 +1,85 @@ +# Strands MCP Server Tests + +This directory contains tests for the Strands MCP Server prompt generation functionality. + +## Test Structure + +- `test_prompts.py` - Unit tests for prompt generation functions +- `test_mcp_server.py` - Integration tests for MCP server endpoints + +## Running Tests + +### Install Dependencies + +First, install the development dependencies: + +```bash +pip install -e ".[dev]" +``` + +### Run All Tests + +```bash +pytest tests -v +``` + +### Run Specific Test Files + +```bash +# Test prompt generation +pytest tests/test_prompts.py -v + +# Test MCP server integration +pytest tests/test_mcp_server.py -v +``` + +### Run with Coverage + +```bash +pytest tests --cov=strands_mcp_server --cov-report=html +``` + +### Run Specific Test Classes or Functions + +```bash +# Run a specific test class +pytest tests/test_prompts.py::TestGenerateToolPrompt -v + +# Run a specific test function +pytest tests/test_mcp_server.py::TestTemplateFiles::test_template_files_exist -v +``` + +## Test Categories + +### Unit Tests +- `TestRegexReplaceFilter` - Tests for the Jinja2 regex filter +- `TestFetchContent` - Tests for content fetching +- `TestGenerateToolPrompt` - Tests for tool prompt generation +- `TestGenerateAgentPrompt` - Tests for agent prompt generation +- `TestGenerateSessionPrompt` - Tests for session prompt generation +- `TestGenerateModelPrompt` - Tests for model provider prompt generation +- `TestGenerateMultiagentPrompt` - Tests for multi-agent prompt generation + +### Integration Tests +- `TestMCPPromptEndpoints` - Tests for MCP server prompt endpoints +- `TestSearchDocsIntegration` - Tests for document search +- `TestFetchDocIntegration` - Tests for document fetching +- `TestMCPServerRegistration` - Tests for proper MCP registration +- `TestTemplateFiles` - Tests for template file existence and validity + +## Adding New Tests + +When adding new prompt templates or functionality: + +1. Add unit tests for the prompt generation function in `test_prompts.py` +2. Add integration tests for the MCP endpoint in `test_mcp_server.py` +3. Ensure the template file exists and has valid Jinja2 syntax +4. Test with both minimal and full parameters + +## Continuous Integration + +These tests should be run as part of the CI/CD pipeline to ensure: +- All templates are present and valid +- Prompt generation works correctly +- MCP endpoints are properly registered +- No regressions in functionality \ No newline at end of file diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..5c61355 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Test suite for Strands MCP Server.""" diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py new file mode 100644 index 0000000..a6573bd --- /dev/null +++ b/tests/test_mcp_server.py @@ -0,0 +1,278 @@ +"""Tests for Strands MCP Server prompt endpoints.""" + +import sys +from pathlib import Path +from unittest.mock import Mock, patch + +import pytest + +# Add src to path for imports +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +from strands_mcp_server.server import ( + fetch_doc, + search_docs, + strands_agent_development, + strands_model_development, + strands_multiagent_development, + strands_session_management, + strands_tool_development, +) + + +class TestMCPPromptEndpoints: + """Test MCP server prompt endpoints.""" + + def test_strands_tool_development_endpoint(self): + """Test the tool development prompt endpoint.""" + result = strands_tool_development( + request="Create a weather API tool", + tool_use_examples="agent.tool.get_weather('London')", + preferred_libraries="requests", + ) + + assert isinstance(result, str) + assert "Create a weather API tool" in result + assert "tool" in result.lower() + assert "@tool" in result + + def test_strands_agent_development_endpoint(self): + """Test the agent development prompt endpoint.""" + result = strands_agent_development( + use_case="Customer support chatbot", + examples="User: Help with order\nAgent: I can help...", + agent_guidelines="Be polite and helpful", + tools_required="database_query, send_email", + model_preferences="Fast response model", + ) + + assert isinstance(result, str) + assert "Customer support chatbot" in result + assert "agent" in result.lower() + assert "system" in result.lower() + + def test_strands_session_management_endpoint(self): + """Test the session management prompt endpoint.""" + result = strands_session_management(request="Add Redis-based session storage", include_examples=True) + + assert isinstance(result, str) + assert "Redis-based session storage" in result + assert "session" in result.lower() + + def test_strands_model_development_endpoint(self): + """Test the model development prompt endpoint.""" + result = strands_model_development( + use_case="Integrate Hugging Face models", + model_details="Support for text generation models", + api_documentation="https://huggingface.co/docs", + auth_requirements="API token", + special_features="Streaming and batching", + ) + + assert isinstance(result, str) + assert "Integrate Hugging Face models" in result + assert "model" in result.lower() + assert "provider" in result.lower() + + def test_strands_multiagent_development_endpoint(self): + """Test the multi-agent development prompt endpoint.""" + result = strands_multiagent_development( + use_case="Research and analysis pipeline", + pattern="graph", + agent_roles="Researcher, Analyst, Writer", + interaction_requirements="Pass context between agents", + scale_requirements="Process 100 documents per hour", + ) + + assert isinstance(result, str) + assert "Research and analysis pipeline" in result + assert "multi-agent" in result.lower() or "multiagent" in result.lower() + assert "graph" in result.lower() or "swarm" in result.lower() + + def test_minimal_parameters(self): + """Test endpoints with minimal parameters.""" + # Tool development + tool_result = strands_tool_development(request="Simple tool") + assert "Simple tool" in tool_result + + # Agent development + agent_result = strands_agent_development(use_case="Simple agent") + assert "Simple agent" in agent_result + + # Session management + session_result = strands_session_management(request="Simple session") + assert "Simple session" in session_result + + # Model development + model_result = strands_model_development(use_case="Simple model") + assert "Simple model" in model_result + + # Multi-agent development + multiagent_result = strands_multiagent_development(use_case="Simple system") + assert "Simple system" in multiagent_result + + +class TestSearchDocsIntegration: + """Test search_docs function integration.""" + + @patch("strands_mcp_server.server.cache") + def test_search_docs_basic(self, mock_cache): + """Test basic document search.""" + # Mock the index and search results + mock_doc = Mock() + mock_doc.uri = "https://example.com/doc1" + mock_doc.display_title = "Test Document" + + mock_index = Mock() + mock_index.search.return_value = [(0.9, mock_doc)] + + mock_cache.get_index.return_value = mock_index + mock_cache.get_url_cache.return_value = {} + mock_cache.SNIPPET_HYDRATE_MAX = 10 # Add this attribute + + result = search_docs("test query", k=5) + + assert isinstance(result, list) + assert len(result) <= 5 + mock_cache.ensure_ready.assert_called_once() + + @patch("strands_mcp_server.server.cache") + def test_search_docs_no_results(self, mock_cache): + """Test search with no results.""" + mock_index = Mock() + mock_index.search.return_value = [] + + mock_cache.get_index.return_value = mock_index + mock_cache.get_url_cache.return_value = {} + mock_cache.SNIPPET_HYDRATE_MAX = 10 # Add this attribute + + result = search_docs("nonexistent", k=5) + + assert result == [] + + +class TestFetchDocIntegration: + """Test fetch_doc function integration.""" + + @patch("strands_mcp_server.server.cache") + def test_fetch_doc_success(self, mock_cache): + """Test successful document fetching.""" + mock_page = Mock() + mock_page.title = "Test Title" + mock_page.content = "Test content" + + mock_cache.ensure_page.return_value = mock_page + + result = fetch_doc("https://example.com/doc") + + assert result["url"] == "https://example.com/doc" + assert result["title"] == "Test Title" + assert result["content"] == "Test content" + + @patch("strands_mcp_server.server.cache") + def test_fetch_doc_failure(self, mock_cache): + """Test document fetch failure.""" + mock_cache.ensure_page.return_value = None + + result = fetch_doc("https://example.com/missing") + + assert result["error"] == "fetch failed" + assert result["url"] == "https://example.com/missing" + + def test_fetch_doc_invalid_uri(self): + """Test with invalid URI.""" + result = fetch_doc("invalid://uri") + + assert result["error"] == "unsupported uri" + assert result["url"] == "invalid://uri" + + +class TestMCPServerRegistration: + """Test that all prompt functions are properly registered with MCP.""" + + def test_prompt_functions_registered(self): + """Verify all prompt functions are registered as MCP prompts.""" + # These should be registered via @mcp.prompt() decorator + expected_prompts = [ + "strands_tool_development", + "strands_agent_development", + "strands_session_management", + "strands_model_development", + "strands_multiagent_development", + ] + + # Check that functions exist and are callable + from strands_mcp_server import server + + for prompt_name in expected_prompts: + assert hasattr(server, prompt_name) + assert callable(getattr(server, prompt_name)) + + def test_tool_functions_registered(self): + """Verify tool functions are registered.""" + expected_tools = ["search_docs", "fetch_doc"] + + from strands_mcp_server import server + + for tool_name in expected_tools: + assert hasattr(server, tool_name) + assert callable(getattr(server, tool_name)) + + +# Test for template file existence +class TestTemplateFiles: + """Test that all required template files exist.""" + + def test_template_files_exist(self): + """Verify all Jinja2 template files exist.""" + template_dir = Path(__file__).parent.parent / "src" / "strands_mcp_server" / "prompts" + + required_templates = [ + "base.jinja2", + "tool_development.jinja2", + "agent_development.jinja2", + "session_management.jinja2", + "model_development.jinja2", + "multiagent_development.jinja2", + ] + + for template in required_templates: + template_path = template_dir / template + assert template_path.exists(), f"Template {template} not found at {template_path}" + + def test_template_syntax_valid(self): + """Test that all templates have valid Jinja2 syntax.""" + import re + + from jinja2 import Environment, FileSystemLoader, TemplateSyntaxError + + template_dir = Path(__file__).parent.parent / "src" / "strands_mcp_server" / "prompts" + env = Environment(loader=FileSystemLoader(template_dir)) + + # Add the regex_replace filter that the templates use + def regex_replace_filter(text, pattern, replacement=""): + if not text: + return text + return re.sub(pattern, replacement, text, flags=re.DOTALL) + + env.filters["regex_replace"] = regex_replace_filter + + templates = [ + "tool_development.jinja2", + "agent_development.jinja2", + "session_management.jinja2", + "model_development.jinja2", + "multiagent_development.jinja2", + ] + + for template_name in templates: + try: + template = env.get_template(template_name) + # Try to render with empty context to check syntax + template.render() + except TemplateSyntaxError as e: + pytest.fail(f"Template {template_name} has syntax error: {e}") + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) diff --git a/tests/test_prompts.py b/tests/test_prompts.py new file mode 100644 index 0000000..134bb88 --- /dev/null +++ b/tests/test_prompts.py @@ -0,0 +1,325 @@ +"""Tests for Strands MCP Server prompt generation functionality.""" + +import sys +from pathlib import Path +from unittest.mock import Mock, patch + +import pytest + +# Add src to path for imports +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +from strands_mcp_server.prompts import ( + fetch_content, + generate_agent_prompt, + generate_model_prompt, + generate_multiagent_prompt, + generate_session_prompt, + generate_tool_prompt, + regex_replace_filter, +) + + +class TestRegexReplaceFilter: + """Test the custom Jinja2 regex replace filter.""" + + def test_regex_replace_basic(self): + """Test basic regex replacement.""" + text = "Hello World 123" + result = regex_replace_filter(text, r"\d+", "XXX") + assert result == "Hello World XXX" + + def test_regex_replace_empty_text(self): + """Test with empty text.""" + result = regex_replace_filter("", r"\d+", "XXX") + assert result == "" + + def test_regex_replace_none_text(self): + """Test with None text.""" + result = regex_replace_filter(None, r"\d+", "XXX") + assert result is None + + def test_regex_replace_multiline(self): + """Test multiline replacement with DOTALL flag.""" + text = "Start\n```\ncode block\n```\nEnd" + result = regex_replace_filter(text, r"```.*?```", "") + assert result == "Start\n\nEnd" + + +class TestFetchContent: + """Test the fetch_content function.""" + + @patch("strands_mcp_server.prompts.cache") + def test_fetch_content_success(self, mock_cache): + """Test successful content fetching.""" + mock_page = Mock() + mock_page.content = "Test content" + mock_cache.ensure_page.return_value = mock_page + + result = fetch_content("https://example.com") + + assert result == "Test content" + mock_cache.ensure_ready.assert_called_once() + mock_cache.ensure_page.assert_called_once_with("https://example.com") + + @patch("strands_mcp_server.prompts.cache") + def test_fetch_content_no_page(self, mock_cache): + """Test when page is not found.""" + mock_cache.ensure_page.return_value = None + + result = fetch_content("https://example.com") + + assert result == "" + + @patch("strands_mcp_server.prompts.cache") + def test_fetch_content_no_content(self, mock_cache): + """Test when page exists but has no content.""" + mock_page = Mock() + mock_page.content = None + mock_cache.ensure_page.return_value = mock_page + + result = fetch_content("https://example.com") + + assert result == "" + + +class TestGenerateToolPrompt: + """Test the generate_tool_prompt function.""" + + @patch("strands_mcp_server.prompts.fetch_content") + @patch("strands_mcp_server.prompts.cache") + def test_generate_tool_prompt_basic(self, mock_cache, mock_fetch): + """Test basic tool prompt generation.""" + mock_fetch.side_effect = ["llms.txt content", "python tools content"] + + result = generate_tool_prompt( + request="Create a calculator tool", + tool_use_examples="agent.tool.calculator(expression='2+2')", + preferred_libraries="sympy", + ) + + assert "Create a calculator tool" in result + assert "agent.tool.calculator" in result + assert "sympy" in result + assert "PHASE 1: DESIGN" in result + assert "PHASE 2: IMPLEMENTATION" in result + mock_cache.ensure_ready.assert_called_once() + + @patch("strands_mcp_server.prompts.fetch_content") + @patch("strands_mcp_server.prompts.cache") + def test_generate_tool_prompt_minimal(self, mock_cache, mock_fetch): + """Test tool prompt with minimal parameters.""" + mock_fetch.side_effect = ["", ""] + + result = generate_tool_prompt(request="Create a tool") + + assert "Create a tool" in result + assert "@tool" in result + assert "Strands" in result + + +class TestGenerateAgentPrompt: + """Test the generate_agent_prompt function.""" + + @patch("strands_mcp_server.prompts.fetch_content") + @patch("strands_mcp_server.prompts.cache") + def test_generate_agent_prompt_full(self, mock_cache, mock_fetch): + """Test agent prompt with all parameters.""" + mock_fetch.side_effect = [ + "llms.txt content", + "agent loop content", + "agent api content", + "community tools content", + ] + + result = generate_agent_prompt( + use_case="Research assistant for academic papers", + examples="User: Find papers on AI\nAgent: Searching...", + agent_guidelines="Be thorough and cite sources", + tools_required="retrieve, file_write", + model_preferences="Claude Sonnet", + include_examples=True, + verbosity="normal", + ) + + assert "Research assistant for academic papers" in result + assert "Find papers on AI" in result + assert "Be thorough and cite sources" in result + assert "retrieve, file_write" in result + assert "Claude Sonnet" in result + assert "PHASE 1: DESIGN" in result + assert "System Prompt" in result + + @patch("strands_mcp_server.prompts.fetch_content") + @patch("strands_mcp_server.prompts.cache") + def test_generate_agent_prompt_minimal(self, mock_cache, mock_fetch): + """Test agent prompt with minimal parameters.""" + mock_fetch.side_effect = ["", "", "", ""] + + result = generate_agent_prompt(use_case="Simple agent") + + assert "Simple agent" in result + assert "Agent" in result + assert "Strands" in result + + +class TestGenerateSessionPrompt: + """Test the generate_session_prompt function.""" + + @patch("strands_mcp_server.prompts.fetch_content") + @patch("strands_mcp_server.prompts.cache") + def test_generate_session_prompt_with_examples(self, mock_cache, mock_fetch): + """Test session prompt with examples included.""" + mock_fetch.side_effect = ["llms.txt content", "session management content", "session api content"] + + result = generate_session_prompt(request="Implement persistent sessions with Redis", include_examples=True) + + assert "Implement persistent sessions with Redis" in result + assert "Session" in result + assert "PHASE 1: DESIGN" in result + assert "Storage" in result + + @patch("strands_mcp_server.prompts.fetch_content") + @patch("strands_mcp_server.prompts.cache") + def test_generate_session_prompt_no_examples(self, mock_cache, mock_fetch): + """Test session prompt without examples.""" + mock_fetch.side_effect = ["", "", ""] + + result = generate_session_prompt(request="Basic session", include_examples=False) + + assert "Basic session" in result + assert "Session Management" in result + + +class TestGenerateModelPrompt: + """Test the generate_model_prompt function.""" + + @patch("strands_mcp_server.prompts.fetch_content") + @patch("strands_mcp_server.prompts.cache") + def test_generate_model_prompt_full(self, mock_cache, mock_fetch): + """Test model provider prompt with all parameters.""" + mock_fetch.side_effect = ["custom model content", "models api content"] + + result = generate_model_prompt( + use_case="Integrate company LLM API", + model_details="GPT-4 compatible endpoint", + api_documentation="https://api.company.com/docs", + auth_requirements="Bearer token authentication", + special_features="Streaming support required", + include_examples=True, + ) + + assert "Integrate company LLM API" in result + assert "GPT-4 compatible endpoint" in result + assert "https://api.company.com/docs" in result + assert "Bearer token authentication" in result + assert "Streaming support required" in result + assert "PHASE 1: DESIGN" in result + assert "Model" in result + + @patch("strands_mcp_server.prompts.fetch_content") + @patch("strands_mcp_server.prompts.cache") + def test_generate_model_prompt_minimal(self, mock_cache, mock_fetch): + """Test model provider prompt with minimal parameters.""" + mock_fetch.side_effect = ["", ""] + + result = generate_model_prompt(use_case="Custom model") + + assert "Custom model" in result + assert "Model Provider" in result + + +class TestGenerateMultiagentPrompt: + """Test the generate_multiagent_prompt function.""" + + @patch("strands_mcp_server.prompts.fetch_content") + @patch("strands_mcp_server.prompts.cache") + def test_generate_multiagent_prompt_graph(self, mock_cache, mock_fetch): + """Test multi-agent prompt for Graph pattern.""" + mock_fetch.side_effect = ["graph content", "swarm content", "multiagent api content"] + + result = generate_multiagent_prompt( + use_case="Document processing pipeline", + pattern="graph", + agent_roles="Processor, Analyzer, Reporter", + interaction_requirements="Sequential processing", + scale_requirements="Handle 1000 docs/hour", + include_examples=True, + ) + + assert "Document processing pipeline" in result + assert "graph" in result.lower() + assert "Processor, Analyzer, Reporter" in result + assert "Sequential processing" in result + assert "Handle 1000 docs/hour" in result + assert "Multi-Agent" in result + + @patch("strands_mcp_server.prompts.fetch_content") + @patch("strands_mcp_server.prompts.cache") + def test_generate_multiagent_prompt_swarm(self, mock_cache, mock_fetch): + """Test multi-agent prompt for Swarm pattern.""" + mock_fetch.side_effect = ["", "", ""] + + result = generate_multiagent_prompt(use_case="Creative problem solving", pattern="swarm") + + assert "Creative problem solving" in result + assert "swarm" in result.lower() + assert "Multi-Agent" in result + + @patch("strands_mcp_server.prompts.fetch_content") + @patch("strands_mcp_server.prompts.cache") + def test_generate_multiagent_prompt_hybrid(self, mock_cache, mock_fetch): + """Test multi-agent prompt for Hybrid pattern.""" + mock_fetch.side_effect = ["", "", ""] + + result = generate_multiagent_prompt(use_case="Complex analysis system", pattern="hybrid") + + assert "Complex analysis system" in result + assert "hybrid" in result.lower() + assert "Multi-Agent" in result + + +# Integration tests +class TestPromptIntegration: + """Integration tests for prompt generation.""" + + @patch("strands_mcp_server.prompts.cache") + def test_all_prompts_generate_without_error(self, mock_cache): + """Test that all prompt generators work without errors.""" + mock_cache.ensure_ready.return_value = None + mock_cache.ensure_page.return_value = None + + # Test each generator + tool_prompt = generate_tool_prompt("Create tool") + assert tool_prompt + assert len(tool_prompt) > 100 + + agent_prompt = generate_agent_prompt("Create agent") + assert agent_prompt + assert len(agent_prompt) > 100 + + session_prompt = generate_session_prompt("Session management") + assert session_prompt + assert len(session_prompt) > 100 + + model_prompt = generate_model_prompt("Custom model") + assert model_prompt + assert len(model_prompt) > 100 + + multiagent_prompt = generate_multiagent_prompt("Multi-agent system") + assert multiagent_prompt + assert len(multiagent_prompt) > 100 + + @patch("strands_mcp_server.prompts.jinja_env") + def test_template_loading_error_handling(self, mock_jinja_env): + """Test handling of template loading errors.""" + mock_jinja_env.get_template.side_effect = Exception("Template not found") + + with pytest.raises(Exception) as exc_info: + generate_tool_prompt("Test") + + assert "Template not found" in str(exc_info.value) + + +if __name__ == "__main__": + pytest.main([__file__, "-v"])