"""Main FastMCP server for AgilePlace integration."""
import logging
import os
from typing import Any, Optional
from fastmcp import FastMCP
from agileplace_mcp.auth import AgilePlaceAuth, AgilePlaceAuthError
from agileplace_mcp.client import AgilePlaceAPIError, AgilePlaceClient, RateLimitError
# Set up logging
logging.basicConfig(
level=os.getenv("LOG_LEVEL", "INFO"),
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
# Initialize FastMCP server first
mcp = FastMCP("AgilePlace MCP Server")
# Now import tools after mcp is initialized
from agileplace_mcp.tools import boards, bulk, cards, connections, dependencies, okr, okr_activities, query
def get_client() -> AgilePlaceClient:
"""Get authenticated AgilePlace client."""
try:
auth = AgilePlaceAuth()
return AgilePlaceClient(auth=auth)
except AgilePlaceAuthError as e:
raise ValueError(f"Authentication failed: {e}")
def handle_api_error(error: Exception) -> str:
"""Handle API errors and return user-friendly messages."""
if isinstance(error, AgilePlaceAPIError):
return f"API Error: {error.message}"
elif isinstance(error, RateLimitError):
return f"Rate limit exceeded. Please wait before trying again."
else:
return f"Unexpected error: {str(error)}"
# Board Management Tools
@mcp.tool()
async def list_boards(
search: Optional[str] = None,
limit: int = 200,
archived: bool = False,
) -> dict:
"""List all boards accessible to the authenticated user."""
try:
client = get_client()
return await boards.list_boards(client, search, limit, archived)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def get_board(
board_id: str,
) -> dict:
"""Get detailed information about a specific board including lanes, card types, and custom fields."""
try:
client = get_client()
return await boards.get_board(client, board_id)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def get_board_cards(
board_id: str,
lanes_json: Optional[str] = None,
limit: int = 200,
) -> dict:
"""Get card faces (summary information) for cards on a board."""
try:
client = get_client()
return await boards.get_board_cards(client, board_id, lanes_json, limit)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def get_leaf_lanes(
board_id: str,
) -> dict:
"""Get lanes that can hold cards (leaf lanes without children)."""
try:
client = get_client()
return await boards.get_leaf_lanes(client, board_id)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def create_board(
title: str,
description: Optional[str] = None,
template_id: Optional[str] = None,
) -> dict:
"""Create a new board."""
try:
client = get_client()
return await boards.create_board(client, title, description, template_id)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def get_board_members(
board_id: str,
search: Optional[str] = None,
) -> dict:
"""Get assigned members (users and teams) on a board."""
try:
client = get_client()
return await boards.get_board_members(client, board_id, search)
except Exception as e:
raise ValueError(handle_api_error(e))
# Card Management Tools
@mcp.tool()
async def list_cards(
board_id: Optional[str] = None,
since: Optional[str] = None,
limit: int = 200,
) -> dict:
"""List cards with optional filtering."""
try:
client = get_client()
return await cards.list_cards(client, board_id, since, limit)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def get_card(
card_id: str,
) -> dict:
"""Get full details of a specific card."""
try:
client = get_client()
return await cards.get_card(client, card_id)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def get_card_activity(
card_id: str,
limit: int = 100,
) -> dict:
"""Get activity history for a card."""
try:
client = get_client()
return await cards.get_card_activity(client, card_id, limit)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def create_card(
board_id: str,
lane_id: str,
title: str,
description: Optional[str] = None,
card_type_id: Optional[str] = None,
priority: Optional[str] = None,
size: Optional[int] = None,
tags_json: Optional[str] = None,
assigned_user_ids_json: Optional[str] = None,
assigned_team_ids_json: Optional[str] = None,
external_card_id: Optional[str] = None,
external_url: Optional[str] = None,
planned_start: Optional[str] = None,
planned_finish: Optional[str] = None,
custom_icon_id: Optional[str] = None,
custom_fields_json: Optional[str] = None,
index: Optional[int] = None,
) -> dict:
"""Create a new card on a board."""
try:
client = get_client()
# Parse JSON parameters
import json
tags = json.loads(tags_json) if tags_json else []
assigned_user_ids = json.loads(assigned_user_ids_json) if assigned_user_ids_json else []
assigned_team_ids = json.loads(assigned_team_ids_json) if assigned_team_ids_json else []
custom_fields = json.loads(custom_fields_json) if custom_fields_json else {}
return await cards.create_card(
client,
board_id,
lane_id,
title,
description,
card_type_id,
priority,
size,
tags,
assigned_user_ids,
assigned_team_ids,
external_card_id,
external_url,
planned_start,
planned_finish,
custom_icon_id,
custom_fields,
index,
)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def update_card(
card_id: str,
title: Optional[str] = None,
description: Optional[str] = None,
priority: Optional[str] = None,
size: Optional[int] = None,
tags_json: Optional[str] = None,
lane_id: Optional[str] = None,
position: Optional[int] = None,
) -> dict:
"""Update fields on an existing card."""
try:
client = get_client()
# Parse JSON parameters
import json
tags = json.loads(tags_json) if tags_json else []
return await cards.update_card(
client,
card_id,
title,
description,
priority,
size,
tags,
lane_id,
position,
)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def move_card(
card_id: str,
lane_id: str,
position: Optional[int] = None,
) -> dict:
"""Move a card to a different lane."""
try:
client = get_client()
return await cards.move_card(client, card_id, lane_id, position)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def delete_card(
card_id: str,
) -> dict:
"""Delete a card."""
try:
client = get_client()
return await cards.delete_card(client, card_id)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def get_card_comments(
card_id: str,
) -> dict:
"""Get comments on a card."""
try:
client = get_client()
return await cards.get_card_comments(client, card_id)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def create_comment(
card_id: str,
text: str,
) -> dict:
"""Add a comment to a card."""
try:
client = get_client()
return await cards.create_comment(client, card_id, text)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def assign_users_to_card(
card_id: str,
user_ids_json: Optional[str] = None,
team_ids_json: Optional[str] = None,
) -> dict:
"""Assign users and/or teams to a card."""
try:
client = get_client()
# Parse JSON parameters
import json
user_ids = json.loads(user_ids_json) if user_ids_json else []
team_ids = json.loads(team_ids_json) if team_ids_json else []
return await cards.assign_users_to_card(
client,
card_id,
user_ids,
team_ids,
)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def get_card_children(
card_id: str,
limit: int = 200,
) -> dict:
"""Get child cards connected to a parent card."""
try:
client = get_client()
return await cards.get_card_children(client, card_id, limit)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def get_card_parents(
card_id: str,
limit: int = 200,
) -> dict:
"""Get parent cards connected to a child card."""
try:
client = get_client()
return await cards.get_card_parents(client, card_id, limit)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def create_connection(
parent_id: str,
child_id: str,
) -> dict:
"""Create a parent-child connection between two cards."""
try:
client = get_client()
return await connections.create_connection(client, parent_id, child_id)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def delete_connection(
parent_id: str,
child_id: str,
) -> dict:
"""Remove a parent-child connection between two cards."""
try:
client = get_client()
return await connections.delete_connection(client, parent_id, child_id)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def get_connection_statistics(
card_id: str,
) -> dict:
"""Get statistics about connected cards (children)."""
try:
client = get_client()
return await connections.get_connection_statistics(client, card_id)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def connect_cards_bulk(
connections_json: str,
) -> dict:
"""Create multiple parent-child connections in a single request."""
try:
client = get_client()
return await connections.connect_cards_bulk(client, connections_json)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def get_card_dependencies(
card_id: str,
) -> dict:
"""Get all dependencies for a card."""
try:
client = get_client()
return await dependencies.get_card_dependencies(client, card_id)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def create_dependency(
card_id: str,
depends_on_card_id: str,
dependency_type: str = "finish_to_start",
) -> dict:
"""Create a dependency between two cards."""
try:
client = get_client()
return await dependencies.create_dependency(
client, card_id, depends_on_card_id, dependency_type
)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def delete_dependency(
dependency_id: str,
) -> dict:
"""Delete a dependency."""
try:
client = get_client()
return await dependencies.delete_dependency(client, dependency_id)
except Exception as e:
raise ValueError(handle_api_error(e))
# Bulk Operations
@mcp.tool()
async def update_cards_bulk(
card_ids_json: str,
updates_json: str,
) -> dict:
"""Update multiple cards with the same field values."""
try:
client = get_client()
return await bulk.update_cards_bulk(client, card_ids_json, updates_json)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def delete_cards_bulk(
card_ids: str,
) -> dict:
"""Delete multiple cards in a single request."""
try:
client = get_client()
return await bulk.delete_cards_bulk(client, card_ids)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def move_cards_bulk(
moves_json: str,
) -> dict:
"""Move multiple cards to different lanes in a single request."""
try:
client = get_client()
return await bulk.move_cards_bulk(client, moves_json)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def assign_members_bulk(
board_ids_json: str,
user_ids_json: Optional[str] = None,
team_ids_json: Optional[str] = None,
board_role: str = "boardUser",
) -> dict:
"""Assign users or teams to multiple boards with a specific role."""
try:
client = get_client()
# Parse JSON parameters
import json
user_ids = json.loads(user_ids_json) if user_ids_json else []
team_ids = json.loads(team_ids_json) if team_ids_json else []
return await bulk.assign_members_bulk(
client,
board_ids_json,
user_ids,
team_ids,
board_role,
)
except Exception as e:
raise ValueError(handle_api_error(e))
# OKR Management Tools
@mcp.tool()
async def create_objective(
external_id: str,
external_type: str,
name: str,
level_depth: int,
starts_at: str,
ends_at: str,
owned_by: str,
parent_objective_id: Optional[int] = None,
description: Optional[str] = None,
ca_values_json: Optional[str] = None,
app_owned_by: Optional[str] = None,
external_title: Optional[str] = None,
) -> dict:
"""Create a new OKR Objective."""
try:
# Parse JSON parameters
import json
ca_values = json.loads(ca_values_json) if ca_values_json else None
return await okr.create_objective(
external_id,
external_type,
name,
level_depth,
starts_at,
ends_at,
owned_by,
parent_objective_id,
description,
ca_values,
app_owned_by,
external_title,
)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def create_key_result(
objective_id: int,
name: str,
owned_by: str,
starts_at: Optional[str] = None,
ends_at: Optional[str] = None,
starting_value: Optional[float] = None,
target_value: Optional[float] = None,
description: Optional[str] = None,
progress_percentage: Optional[int] = None,
) -> dict:
"""Create a new Key Result for an Objective."""
try:
return await okr.create_key_result(
objective_id,
name,
owned_by,
starts_at,
ends_at,
starting_value,
target_value,
description,
progress_percentage,
)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def update_objective(
objective_id: int,
name: Optional[str] = None,
description: Optional[str] = None,
starts_at: Optional[str] = None,
ends_at: Optional[str] = None,
ca_values_json: Optional[str] = None,
) -> dict:
"""Update an existing OKR Objective."""
try:
# Parse JSON parameters
import json
ca_values = json.loads(ca_values_json) if ca_values_json else None
return await okr.update_objective(
objective_id,
name,
description,
starts_at,
ends_at,
ca_values,
)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def update_key_result(
key_result_id: int,
name: Optional[str] = None,
description: Optional[str] = None,
starts_at: Optional[str] = None,
ends_at: Optional[str] = None,
starting_value: Optional[float] = None,
target_value: Optional[float] = None,
progress_percentage: Optional[int] = None,
) -> dict:
"""Update an existing Key Result."""
try:
return await okr.update_key_result(
key_result_id,
name,
description,
starts_at,
ends_at,
starting_value,
target_value,
progress_percentage,
)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def get_objectives_by_board_id(
board_id: str,
level_depth: Optional[int] = None,
) -> dict:
"""Get OKR Objectives for a specific board."""
try:
return await okr.get_objectives_by_board_id(board_id, level_depth)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def get_key_results_by_objective_id(
objective_id: int,
) -> dict:
"""Get Key Results for a specific Objective."""
try:
return await okr.get_key_results_by_objective_id(objective_id)
except Exception as e:
raise ValueError(handle_api_error(e))
# Additional OKR Management Tools
@mcp.tool()
async def get_objective_by_id(
objective_id: int,
) -> dict:
"""Get a specific Objective by ID."""
try:
return await okr.get_objective_by_id(objective_id)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def get_key_result_by_id(
key_result_id: int,
) -> dict:
"""Get a specific Key Result by ID."""
try:
return await okr.get_key_result_by_id(key_result_id)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def delete_objective(
objective_id: int,
) -> dict:
"""Delete an Objective."""
try:
return await okr.delete_objective(objective_id)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def delete_key_result(
key_result_id: int,
) -> dict:
"""Delete a Key Result."""
try:
return await okr.delete_key_result(key_result_id)
except Exception as e:
raise ValueError(handle_api_error(e))
# OKR Activity and Work Item Management Tools
@mcp.tool()
async def connect_activities_to_key_result(
key_result_id: int,
work_item_container_external_id: str,
work_item_container_external_type: str,
work_items_json: str,
) -> dict:
"""Connect a set of work items (cards) to a key result."""
try:
client = get_client()
# Parse JSON parameters
import json
work_items = json.loads(work_items_json)
# Create work item container
work_item_container = okr_activities.WorkItemContainer(
external_id=work_item_container_external_id,
external_type=work_item_container_external_type,
)
# Create work items
work_item_objects = []
for item in work_items:
work_item_objects.append(
okr_activities.WorkItem(
external_id=item["external_id"],
external_type=item["external_type"],
title=item.get("title"),
state=item.get("state"),
)
)
return await okr_activities.connect_activities_to_key_result(
client,
key_result_id,
work_item_container,
work_item_objects,
)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def create_activity(
key_result_id: int,
context_id: str,
context_title: str,
title: str,
external_activity_type_id: str,
product_type: str,
planned_start: Optional[str] = None,
planned_finish: Optional[str] = None,
) -> dict:
"""Create a new activity associated with a key result."""
try:
client = get_client()
return await okr_activities.create_activity(
client,
key_result_id,
context_id,
context_title,
title,
external_activity_type_id,
product_type,
planned_start,
planned_finish,
)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def list_activities(
key_result_id: Optional[int] = None,
work_item_container_id: Optional[int] = None,
activity_ids_json: Optional[str] = None,
search_string: Optional[str] = None,
limit: int = 50,
product_type: Optional[str] = None,
) -> dict:
"""List activities based on various filters."""
try:
client = get_client()
# Parse JSON parameters
import json
activity_ids = json.loads(activity_ids_json) if activity_ids_json else None
return await okr_activities.list_activities(
client,
key_result_id,
work_item_container_id,
activity_ids,
search_string,
limit,
product_type,
)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def list_activity_containers(
container_ids_json: Optional[str] = None,
search_string: Optional[str] = None,
limit: int = 50,
product_types_json: Optional[str] = None,
) -> dict:
"""List activity containers."""
try:
client = get_client()
# Parse JSON parameters
import json
container_ids = json.loads(container_ids_json) if container_ids_json else None
product_types = json.loads(product_types_json) if product_types_json else None
return await okr_activities.list_activity_containers(
client,
container_ids,
search_string,
limit,
product_types,
)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def search_activities(
context_id: str,
product_type: str,
search_string: Optional[str] = None,
limit: int = 50,
) -> dict:
"""Search activities based on context and product type."""
try:
client = get_client()
return await okr_activities.search_activities(
client,
context_id,
product_type,
search_string,
limit,
)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def list_activity_types(
context_id: str,
product_type: str,
) -> dict:
"""List available activity types for a context."""
try:
client = get_client()
return await okr_activities.list_activity_types(
client,
context_id,
product_type,
)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def search_users(
context_id: str,
product_type: str,
search_string: Optional[str] = None,
limit: int = 50,
) -> dict:
"""Search users within a context."""
try:
client = get_client()
return await okr_activities.search_users(
client,
context_id,
product_type,
search_string,
limit,
)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def get_current_user(
context_ids_json: Optional[str] = None,
product_types_json: Optional[str] = None,
) -> dict:
"""Get information about the current user."""
try:
client = get_client()
# Parse JSON parameters
import json
context_ids = json.loads(context_ids_json) if context_ids_json else None
product_types = json.loads(product_types_json) if product_types_json else None
return await okr_activities.get_current_user(
client,
context_ids,
product_types,
)
except Exception as e:
raise ValueError(handle_api_error(e))
# Query Tools
@mcp.tool()
async def list_projects(
portfolio_id: Optional[str] = None,
status: Optional[str] = None,
limit: Optional[int] = None,
attributes: Optional[str] = None,
) -> dict:
"""List projects using the work endpoint with filters."""
try:
client = get_client()
return await query.list_projects(
client,
portfolio_id,
status,
limit,
attributes,
)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def get_project_attributes(
) -> dict:
"""List available project attributes."""
try:
client = get_client()
return await query.get_project_attributes(client)
except Exception as e:
raise ValueError(handle_api_error(e))
@mcp.tool()
async def get_work_attributes(
) -> dict:
"""Get available work attributes."""
try:
client = get_client()
return await query.get_work_attributes(client)
except Exception as e:
raise ValueError(handle_api_error(e))
def main():
"""Main entry point for the server."""
mcp.run()
if __name__ == "__main__":
main()
