diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..c051d07 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,121 @@ +# Webex Bot MCP — Developer Guide + +## Project Overview + +A Model Context Protocol (MCP) server that bridges AI assistants with the Webex Teams API. +It exposes tools for managing rooms, messages, and memberships, plus resources (contextual +guides) and prompt templates for common workflows. + +## Repository Layout + +``` +main.py — MCP server entry point; registers all tools/resources/prompts +config.py — WebexConfig dataclass; loaded via get_config() or WebexConfig.from_env() +health_check.py — Standalone diagnostic script; validates env and API connectivity +pyproject.toml — Project metadata and dependencies (uses uv) +Dockerfile — Multi-stage build; non-root user, health checks +docker-compose.yml — Compose stack with optional Prometheus + Grafana + +tools/ + __init__.py — Re-exports every tool function + common.py — Shared: WebexAPI client, create_error_response, create_success_response, + WebexErrorCodes, version constants + rooms.py — Room/space CRUD (list, create, update, get, delete) + space aliases + messages.py — Message send/list/delete + mention helpers + space aliases + memberships.py — Membership CRUD (list, add, update, delete) + space aliases + people.py — get_webex_me, list_webex_people + +tests/ + test_messages.py — Unit tests (38 cases); mocks webexpythonsdk at sys.modules level +``` + +## Key Conventions + +### Tool count +31 tools total: 5 room + 5 space-room aliases + 4 message + 2 space-message aliases + +4 membership + 2 space-membership aliases + 2 people + 4 space-room/membership aliases. +Update `tools_count` in the `server_version` resource (`main.py`) when adding tools. + +### Error handling — always use structured responses +Every tool must return via `create_error_response` or `create_success_response` from +`tools/common.py`. Never return a bare `{'success': False, 'error': str(e)}` dict. +Map exceptions with a local `_map_exception_to_error(e)` helper (see any tool module for +the pattern). + +Error code ranges: +- `E001–E003` — client/argument errors (do not retry) +- `E401–E404` — auth/access errors (do not retry) +- `E500–E504` — server errors (retry with backoff) +- `E600–E602` — Webex-specific errors + +### Response shape +```python +# Error +{'success': False, 'error_code': 'E001', 'message': '...', 'timestamp': '', 'server_version': '...'} + +# Success +{'success': True, 'data': {...}, 'timestamp': '', 'server_version': '...', 'metadata': {...}} +``` + +### Space aliases +Every room/message/membership tool has a "space" alias that delegates to the room variant +and renames keys in the response (`room` → `space`, `rooms` → `spaces`, etc.). +Aliases live in the same module file as the canonical tool. + +### `files` parameter +Always wrap a string URL in a list before sending to the SDK: +```python +params['files'] = [files] if isinstance(files, str) else files +``` + +## Running the Server + +```bash +# stdio (default — for Claude Desktop / MCP clients) +uv run python main.py + +# HTTP transport +uv run python main.py --transport streamable-http --host 0.0.0.0 --port 8000 +``` + +Required environment variable: `WEBEX_ACCESS_TOKEN` + +## Running Tests + +```bash +python -m unittest discover -s tests -v +``` + +Tests mock `webexpythonsdk` via `sys.modules` so no real credentials are needed. + +## Adding a New Tool + +1. Implement the function in the appropriate `tools/*.py` module. +2. Use `create_error_response` / `create_success_response` for all returns. +3. Add a `_map_exception_to_error` call in the `except` block. +4. Export from `tools/__init__.py`. +5. Register with `mcp.tool()(your_function)` in `main.py`. +6. Update `tools_count` in the `server_version` resource in `main.py`. +7. Add unit tests in `tests/test_messages.py` (or a new test file). + +## Environment Variables + +| Variable | Required | Default | Description | +|---|---|---|---| +| `WEBEX_ACCESS_TOKEN` | ✅ | — | Bot token from developer.webex.com | +| `WEBEX_DEBUG` | | `false` | Enable debug logging | +| `WEBEX_RATE_LIMIT_MESSAGES_PER_SECOND` | | `10` | Message rate limit | +| `WEBEX_RATE_LIMIT_API_CALLS_PER_MINUTE` | | `300` | API rate limit | +| `WEBEX_TIMEOUT_SECONDS` | | `30` | HTTP timeout | +| `LOG_LEVEL` | | `INFO` | `DEBUG`, `INFO`, `WARNING`, or `ERROR` | +| `LOG_FORMAT` | | `text` | `text` or `json` | +| `METRICS_ENABLED` | | `false` | Enable metrics endpoint | +| `METRICS_ENDPOINT` | | — | Prometheus push endpoint | + +## Dependency Management + +Uses `uv`. The declared runtime dependency for dotenv is `python-dotenv>=1.0.0` +(not the unrelated `dotenv` PyPI package). After changing `pyproject.toml` run: +```bash +uv lock && uv sync +``` diff --git a/config.py b/config.py index 98eb3b2..ad5de6b 100644 --- a/config.py +++ b/config.py @@ -103,7 +103,7 @@ def validate(self) -> list[str]: if self.timeout_seconds <= 0: issues.append("Timeout seconds must be positive") - if self.log_level not in ["DEBUG", "INFO", "WARN", "ERROR"]: + if self.log_level not in ["DEBUG", "INFO", "WARNING", "ERROR"]: issues.append("Log level must be one of: DEBUG, INFO, WARN, ERROR") if self.log_format not in ["text", "json"]: diff --git a/main.py b/main.py index 439193e..30b8c20 100644 --- a/main.py +++ b/main.py @@ -21,31 +21,32 @@ # Import all tool functions from the tools package from tools import ( # Room functions - list_webex_rooms, create_webex_room, update_webex_room, get_webex_room, - # Space aliases - list_webex_spaces, create_webex_space, update_webex_space, get_webex_space, + list_webex_rooms, create_webex_room, update_webex_room, + get_webex_room, delete_webex_room, + # Space aliases + list_webex_spaces, create_webex_space, update_webex_space, + get_webex_space, delete_webex_space, # Message functions - send_webex_message, list_webex_messages, send_webex_message_with_mentions, + send_webex_message, send_webex_message_with_mentions, + list_webex_messages, delete_webex_message, # Space message aliases send_webex_space_message, list_webex_space_messages, # Membership functions - list_webex_memberships, add_webex_membership, update_webex_membership, + list_webex_memberships, add_webex_membership, + update_webex_membership, delete_webex_membership, # Space membership aliases list_webex_space_memberships, add_webex_space_membership, # People functions - get_webex_me, list_webex_people + get_webex_me, list_webex_people, ) # Import version and error handling from common from tools.common import MCP_SERVER_VERSION, MCP_SPEC_VERSION -# Load environment variables from .env file +# Load environment variables before importing tools (tools/common.py reads them at import time) load_dotenv() -# Validate required environment variables webex_access_token = os.getenv("WEBEX_ACCESS_TOKEN") -if not webex_access_token: - raise ValueError("WEBEX_ACCESS_TOKEN environment variable is required") # Initialize FastMCP with a name for the bot mcp = FastMCP("Webex Bot MCP") @@ -56,17 +57,20 @@ mcp.tool()(create_webex_room) mcp.tool()(update_webex_room) mcp.tool()(get_webex_room) +mcp.tool()(delete_webex_room) # Space aliases (same functionality as rooms but with "space" terminology) mcp.tool()(list_webex_spaces) mcp.tool()(create_webex_space) mcp.tool()(update_webex_space) mcp.tool()(get_webex_space) +mcp.tool()(delete_webex_space) # Message management tools mcp.tool()(send_webex_message) mcp.tool()(send_webex_message_with_mentions) mcp.tool()(list_webex_messages) +mcp.tool()(delete_webex_message) # Space message aliases mcp.tool()(send_webex_space_message) @@ -76,6 +80,7 @@ mcp.tool()(list_webex_memberships) mcp.tool()(add_webex_membership) mcp.tool()(update_webex_membership) +mcp.tool()(delete_webex_membership) # Space membership aliases mcp.tool()(list_webex_space_memberships) @@ -767,12 +772,12 @@ def server_version(): "mcp_version": MCP_SPEC_VERSION, "api_version": "v1", "supported_features": [ - "tools", "resources", "prompts", + "tools", "resources", "prompts", "streamable-http", "stdio", "error-handling", "versioning" ], - "tools_count": 26, - "resources_count": 8, + "tools_count": 31, + "resources_count": 10, "prompts_count": 7, "breaking_changes": { "1.0.0": [ diff --git a/pyproject.toml b/pyproject.toml index 9047958..315fe51 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,11 +1,11 @@ [project] name = "webex-bot-mcp" -version = "0.1.0" -description = "Add your description here" +version = "1.0.0" +description = "Model Context Protocol server for Webex Teams — manage rooms, messages, and memberships via AI assistants" readme = "README.md" requires-python = ">=3.12" dependencies = [ - "dotenv>=0.9.9", + "python-dotenv>=1.0.0", "fastmcp>=2.9.0", "webexpythonsdk>=2.0.4", ] diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_messages.py b/tests/test_messages.py new file mode 100644 index 0000000..f0023e4 --- /dev/null +++ b/tests/test_messages.py @@ -0,0 +1,341 @@ +""" +Unit tests for message tools and shared utilities. + +webexpythonsdk is mocked at the sys.modules level so tests can run without +real Webex credentials or the SDK being installed. +""" +import os +import sys +import types +import unittest +from unittest.mock import MagicMock, patch + +# ── Provide fake env var and SDK before any tools module is imported ────────── +os.environ.setdefault("WEBEX_ACCESS_TOKEN", "test-token") + +_sdk_mod = types.ModuleType("webexpythonsdk") +_sdk_mod.WebexAPI = MagicMock(return_value=MagicMock()) +sys.modules.setdefault("webexpythonsdk", _sdk_mod) + +# ── Also stub python-dotenv if missing ─────────────────────────────────────── +if "dotenv" not in sys.modules: + _dotenv_mod = types.ModuleType("dotenv") + _dotenv_mod.load_dotenv = lambda *a, **kw: None + sys.modules["dotenv"] = _dotenv_mod + +# Now it's safe to import the tools package +from tools.messages import ( # noqa: E402 + format_mention_by_email, + format_mention_by_person_id, + format_mention_all, + create_message_with_mentions, + send_webex_message, + send_webex_message_with_mentions, + list_webex_messages, +) +from tools.common import ( # noqa: E402 + create_error_response, + create_success_response, + WebexErrorCodes, +) +from config import WebexConfig # noqa: E402 +import tools.messages as _msg_mod # noqa: E402 + + +# ── helpers ─────────────────────────────────────────────────────────────────── + +def _fake_message(**overrides): + m = MagicMock() + m.id = "msg1" + m.roomId = "ROOMID" + m.text = "hello" + m.personId = "pid" + m.personEmail = "bot@example.com" + m.created = "2025-01-01T00:00:00Z" + for k, v in overrides.items(): + setattr(m, k, v) + return m + + +# ── mention formatters ──────────────────────────────────────────────────────── + +class TestMentionFormatters(unittest.TestCase): + def test_email_no_display_name(self): + self.assertEqual( + format_mention_by_email("alice@example.com"), + "<@personEmail:alice@example.com>" + ) + + def test_email_with_display_name(self): + self.assertEqual( + format_mention_by_email("alice@example.com", "Alice"), + "<@personEmail:alice@example.com|Alice>" + ) + + def test_person_id_no_display_name(self): + self.assertEqual( + format_mention_by_person_id("abc123"), + "<@personId:abc123>" + ) + + def test_person_id_with_display_name(self): + self.assertEqual( + format_mention_by_person_id("abc123", "Bob"), + "<@personId:abc123|Bob>" + ) + + def test_mention_all(self): + self.assertEqual(format_mention_all(), "<@all>") + + +# ── create_message_with_mentions ────────────────────────────────────────────── + +class TestCreateMessageWithMentions(unittest.TestCase): + def test_no_mentions_returns_base(self): + self.assertEqual(create_message_with_mentions("hello"), "hello") + + def test_empty_list_returns_base(self): + self.assertEqual(create_message_with_mentions("hello", []), "hello") + + def test_email_mention_prepended(self): + result = create_message_with_mentions( + "check this out", + [{"type": "email", "value": "alice@example.com", "display_name": "Alice"}] + ) + self.assertIn("<@personEmail:alice@example.com|Alice>", result) + self.assertTrue(result.endswith("check this out")) + + def test_all_mention_prepended(self): + result = create_message_with_mentions("hey", [{"type": "all"}]) + self.assertTrue(result.startswith("<@all>")) + self.assertIn("hey", result) + + def test_person_id_mention(self): + result = create_message_with_mentions( + "hi", [{"type": "person_id", "value": "abc123"}] + ) + self.assertIn("<@personId:abc123>", result) + + def test_unknown_type_skipped(self): + result = create_message_with_mentions("hi", [{"type": "unknown", "value": "x"}]) + self.assertEqual(result, "hi") + + def test_empty_email_value_skipped(self): + result = create_message_with_mentions("hi", [{"type": "email", "value": ""}]) + self.assertEqual(result, "hi") + + def test_multiple_mentions_in_order(self): + result = create_message_with_mentions( + "msg", + [ + {"type": "all"}, + {"type": "email", "value": "x@y.com"}, + ] + ) + all_pos = result.index("<@all>") + email_pos = result.index("<@personEmail:x@y.com>") + self.assertLess(all_pos, email_pos) + + +# ── send_webex_message validation ───────────────────────────────────────────── + +class TestSendWebexMessageValidation(unittest.TestCase): + def _send(self, **kwargs): + mock_api = MagicMock() + with patch.object(_msg_mod, 'webex_api', mock_api): + return send_webex_message(**kwargs) + + def test_missing_destination_is_e001(self): + r = self._send(text="hello") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E001') + + def test_missing_content_is_e002(self): + r = self._send(room_id="ROOMID") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E002') + + def test_text_too_long_is_e003(self): + r = self._send(room_id="ROOMID", text="x" * 7440) + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E003') + + def test_markdown_too_long_is_e003(self): + r = self._send(room_id="ROOMID", markdown="x" * 7440) + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E003') + + def test_files_string_wrapped_in_list(self): + mock_api = MagicMock() + mock_api.messages.create.return_value = _fake_message() + with patch.object(_msg_mod, 'webex_api', mock_api): + send_webex_message(room_id="ROOMID", files="https://example.com/f.pdf") + call_kwargs = mock_api.messages.create.call_args[1] + self.assertIsInstance(call_kwargs['files'], list) + self.assertEqual(call_kwargs['files'], ["https://example.com/f.pdf"]) + + def test_success_has_nonempty_timestamp(self): + mock_api = MagicMock() + mock_api.messages.create.return_value = _fake_message() + with patch.object(_msg_mod, 'webex_api', mock_api): + r = send_webex_message(room_id="ROOMID", text="hello") + self.assertTrue(r['success']) + self.assertIn('timestamp', r) + self.assertNotEqual(r['timestamp'], '') + + def test_room_id_routed_to_roomid_param(self): + mock_api = MagicMock() + mock_api.messages.create.return_value = _fake_message() + with patch.object(_msg_mod, 'webex_api', mock_api): + send_webex_message(room_id="ROOMID", text="hi") + self.assertEqual(mock_api.messages.create.call_args[1]['roomId'], "ROOMID") + + def test_person_email_routed_correctly(self): + mock_api = MagicMock() + mock_api.messages.create.return_value = _fake_message() + with patch.object(_msg_mod, 'webex_api', mock_api): + send_webex_message(to_person_email="user@example.com", text="hi") + self.assertEqual( + mock_api.messages.create.call_args[1]['toPersonEmail'], "user@example.com" + ) + + +# ── send_webex_message_with_mentions ───────────────────────────────────────── + +class TestSendWebexMessageWithMentions(unittest.TestCase): + def _send(self, **kwargs): + mock_api = MagicMock() + with patch.object(_msg_mod, 'webex_api', mock_api): + return send_webex_message_with_mentions(**kwargs) + + def test_missing_destination_is_e001(self): + r = self._send(text="hello") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E001') + + def test_missing_content_is_e002(self): + r = self._send(room_id="ROOMID") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E002') + + def test_mention_prepended_to_markdown(self): + mock_api = MagicMock() + mock_api.messages.create.return_value = _fake_message() + with patch.object(_msg_mod, 'webex_api', mock_api): + send_webex_message_with_mentions( + room_id="ROOMID", + markdown="update", + mentions=[{"type": "all"}] + ) + sent_markdown = mock_api.messages.create.call_args[1]['markdown'] + self.assertIn("<@all>", sent_markdown) + self.assertIn("update", sent_markdown) + self.assertLess( + sent_markdown.index("<@all>"), + sent_markdown.index("update") + ) + + +# ── list_webex_messages ─────────────────────────────────────────────────────── + +class TestListWebexMessages(unittest.TestCase): + def test_returns_structured_data(self): + mock_api = MagicMock() + mock_api.messages.list.return_value = [_fake_message()] + with patch.object(_msg_mod, 'webex_api', mock_api): + r = list_webex_messages(room_id="ROOMID") + self.assertTrue(r['success']) + self.assertIn('messages', r['data']) + self.assertEqual(r['metadata']['count'], 1) + + def test_empty_room_returns_empty_list(self): + mock_api = MagicMock() + mock_api.messages.list.return_value = [] + with patch.object(_msg_mod, 'webex_api', mock_api): + r = list_webex_messages(room_id="ROOMID") + self.assertTrue(r['success']) + self.assertEqual(r['data']['messages'], []) + self.assertEqual(r['metadata']['count'], 0) + + +# ── create_error_response / create_success_response ────────────────────────── + +class TestResponseBuilders(unittest.TestCase): + def test_error_has_required_fields(self): + r = create_error_response(WebexErrorCodes.NOT_FOUND, "not found") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E404') + self.assertIn('timestamp', r) + self.assertNotEqual(r['timestamp'], '') + self.assertIn('server_version', r) + + def test_success_has_required_fields(self): + r = create_success_response({'key': 'value'}) + self.assertTrue(r['success']) + self.assertEqual(r['data'], {'key': 'value'}) + self.assertIn('timestamp', r) + self.assertNotEqual(r['timestamp'], '') + + def test_temporary_error_has_retry_info(self): + r = create_error_response( + WebexErrorCodes.RATE_LIMITED, "rate limited", + temporary=True, retry_after_seconds=60 + ) + self.assertTrue(r['temporary']) + self.assertEqual(r['retry_after_seconds'], 60) + + def test_non_temporary_error_has_no_retry(self): + r = create_error_response(WebexErrorCodes.UNAUTHORIZED, "unauthorized") + self.assertNotIn('temporary', r) + self.assertNotIn('retry_after_seconds', r) + + +# ── WebexConfig ─────────────────────────────────────────────────────────────── + +class TestConfig(unittest.TestCase): + def test_valid_config_no_issues(self): + cfg = WebexConfig(access_token="tok") + self.assertEqual(cfg.validate(), []) + + def test_missing_token_fails(self): + cfg = WebexConfig(access_token="") + issues = cfg.validate() + self.assertTrue(any("WEBEX_ACCESS_TOKEN" in i for i in issues)) + + def test_warning_log_level_accepted(self): + cfg = WebexConfig(access_token="tok", log_level="WARNING") + self.assertEqual(cfg.validate(), []) + + def test_warn_log_level_rejected(self): + cfg = WebexConfig(access_token="tok", log_level="WARN") + issues = cfg.validate() + self.assertTrue(any("Log level" in i for i in issues)) + + def test_negative_rate_limit_fails(self): + cfg = WebexConfig(access_token="tok", rate_limit_messages_per_second=-1) + issues = cfg.validate() + self.assertTrue(any("messages per second" in i for i in issues)) + + def test_negative_timeout_fails(self): + cfg = WebexConfig(access_token="tok", timeout_seconds=-1) + issues = cfg.validate() + self.assertTrue(any("Timeout" in i for i in issues)) + + def test_to_dict_does_not_expose_raw_token(self): + cfg = WebexConfig(access_token="supersecret") + d = cfg.to_dict() + self.assertNotIn("supersecret", str(d)) + self.assertTrue(d["access_token_configured"]) + + def test_from_env_raises_if_token_missing(self): + env_backup = os.environ.pop("WEBEX_ACCESS_TOKEN", None) + try: + with self.assertRaises(ValueError): + WebexConfig.from_env() + finally: + if env_backup: + os.environ["WEBEX_ACCESS_TOKEN"] = env_backup + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/__init__.py b/tools/__init__.py index 7174e4a..5ee7131 100644 --- a/tools/__init__.py +++ b/tools/__init__.py @@ -4,39 +4,43 @@ This package contains all the MCP tools organized by functionality area. """ -# Import all tool functions to make them available from the package from .rooms import ( - list_webex_rooms, create_webex_room, update_webex_room, get_webex_room, - list_webex_spaces, create_webex_space, update_webex_space, get_webex_space + list_webex_rooms, create_webex_room, update_webex_room, get_webex_room, delete_webex_room, + list_webex_spaces, create_webex_space, update_webex_space, get_webex_space, delete_webex_space, ) from .messages import ( - send_webex_message, list_webex_messages, send_webex_message_with_mentions, - send_webex_space_message, list_webex_space_messages + send_webex_message, send_webex_message_with_mentions, + list_webex_messages, delete_webex_message, + send_webex_space_message, list_webex_space_messages, ) from .memberships import ( - list_webex_memberships, add_webex_membership, update_webex_membership, - list_webex_space_memberships, add_webex_space_membership + list_webex_memberships, add_webex_membership, update_webex_membership, delete_webex_membership, + list_webex_space_memberships, add_webex_space_membership, ) from .people import ( - get_webex_me, list_webex_people + get_webex_me, list_webex_people, ) __all__ = [ # Room functions - 'list_webex_rooms', 'create_webex_room', 'update_webex_room', 'get_webex_room', + 'list_webex_rooms', 'create_webex_room', 'update_webex_room', + 'get_webex_room', 'delete_webex_room', # Space aliases - 'list_webex_spaces', 'create_webex_space', 'update_webex_space', 'get_webex_space', + 'list_webex_spaces', 'create_webex_space', 'update_webex_space', + 'get_webex_space', 'delete_webex_space', # Message functions - 'send_webex_message', 'list_webex_messages', 'send_webex_message_with_mentions', + 'send_webex_message', 'send_webex_message_with_mentions', + 'list_webex_messages', 'delete_webex_message', # Space message aliases 'send_webex_space_message', 'list_webex_space_messages', # Membership functions - 'list_webex_memberships', 'add_webex_membership', 'update_webex_membership', + 'list_webex_memberships', 'add_webex_membership', + 'update_webex_membership', 'delete_webex_membership', # Space membership aliases 'list_webex_space_memberships', 'add_webex_space_membership', # People functions - 'get_webex_me', 'list_webex_people' + 'get_webex_me', 'list_webex_people', ] diff --git a/tools/common.py b/tools/common.py index 66602db..991d049 100644 --- a/tools/common.py +++ b/tools/common.py @@ -2,8 +2,8 @@ Common utilities and shared components for Webex Bot MCP tools. """ import os +from datetime import datetime, timezone from typing import Dict, Any -from dotenv import load_dotenv from webexpythonsdk import WebexAPI # Server version information @@ -58,7 +58,7 @@ def create_error_response( 'success': False, 'error_code': error_code, 'message': message, - 'timestamp': os.environ.get('REQUEST_TIMESTAMP', ''), + 'timestamp': datetime.now(timezone.utc).isoformat(), 'server_version': MCP_SERVER_VERSION } @@ -88,7 +88,7 @@ def create_success_response(data: Dict[str, Any], metadata: Dict[str, Any] = Non response = { 'success': True, 'data': data, - 'timestamp': os.environ.get('REQUEST_TIMESTAMP', ''), + 'timestamp': datetime.now(timezone.utc).isoformat(), 'server_version': MCP_SERVER_VERSION } @@ -98,10 +98,7 @@ def create_success_response(data: Dict[str, Any], metadata: Dict[str, Any] = Non return response -# Load environment variables from .env file -load_dotenv() - -# Initialize Webex SDK +# Initialize Webex SDK — token must be set before importing this module webex_access_token = os.getenv("WEBEX_ACCESS_TOKEN") if not webex_access_token: raise ValueError("WEBEX_ACCESS_TOKEN environment variable is required") diff --git a/tools/memberships.py b/tools/memberships.py index a5f0e67..a7ca0bf 100644 --- a/tools/memberships.py +++ b/tools/memberships.py @@ -2,7 +2,49 @@ Webex Membership management tools. """ from typing import Optional, Dict, Any -from .common import webex_api +from .common import webex_api, create_error_response, create_success_response, WebexErrorCodes + + +def _map_exception_to_error(e: Exception) -> Dict[str, Any]: + error_str = str(e).lower() + if 'unauthorized' in error_str or 'invalid token' in error_str: + return create_error_response(WebexErrorCodes.UNAUTHORIZED, + "Invalid or expired bot token.") + if 'not found' in error_str or '404' in error_str: + return create_error_response(WebexErrorCodes.NOT_FOUND, + "Membership, room, or person not found.") + if 'forbidden' in error_str or '403' in error_str: + return create_error_response(WebexErrorCodes.FORBIDDEN, + "Bot lacks permission for this operation. " + "Ensure the bot has moderator privileges.") + if 'rate limit' in error_str or 'too many requests' in error_str: + return create_error_response(WebexErrorCodes.RATE_LIMITED, + "API rate limit exceeded. Please retry after delay.", + temporary=True, retry_after_seconds=60) + if 'network' in error_str or 'connection' in error_str: + return create_error_response(WebexErrorCodes.NETWORK_ERROR, + "Network connectivity issue. Please retry.", + temporary=True, retry_after_seconds=30) + return create_error_response(WebexErrorCodes.WEBEX_API_ERROR, + f"Webex API error: {e}", + temporary=True, details={'original_error': str(e)}) + + +def _membership_to_dict(m) -> Dict[str, Any]: + d: Dict[str, Any] = { + 'id': m.id, + 'roomId': m.roomId, + 'personId': m.personId, + 'personEmail': m.personEmail, + 'personDisplayName': m.personDisplayName, + 'personOrgId': m.personOrgId, + 'isModerator': m.isModerator, + 'isMonitor': m.isMonitor, + 'created': m.created, + } + if getattr(m, 'roomType', None): + d['roomType'] = m.roomType + return d def list_webex_memberships( @@ -13,19 +55,18 @@ def list_webex_memberships( ) -> Dict[str, Any]: """ List memberships for a room or person. - + Args: room_id: Room ID to get memberships for person_id: Person ID to get memberships for person_email: Person email to get memberships for max_results: Maximum number of memberships to return (default 100, max 1000) - + Returns: - Dictionary containing the list of memberships and metadata + Standardized response dictionary with success/error information """ try: - # Build parameters dict - params = {} + params: Dict[str, Any] = {} if room_id: params['roomId'] = room_id if person_id: @@ -34,45 +75,14 @@ def list_webex_memberships( params['personEmail'] = person_email if max_results: params['max'] = max_results - - # Call the Webex API - memberships_response = webex_api.memberships.list(**params) - - # Convert the response to a list of dictionaries - memberships_list = [] - for membership in memberships_response: - membership_dict = { - 'id': membership.id, - 'roomId': membership.roomId, - 'personId': membership.personId, - 'personEmail': membership.personEmail, - 'personDisplayName': membership.personDisplayName, - 'personOrgId': membership.personOrgId, - 'isModerator': membership.isModerator, - 'isMonitor': membership.isMonitor, - 'created': membership.created - } - - # Add optional fields if they exist - if hasattr(membership, 'roomType') and membership.roomType: - membership_dict['roomType'] = membership.roomType - - memberships_list.append(membership_dict) - - return { - 'success': True, - 'memberships': memberships_list, - 'count': len(memberships_list), - 'filters_applied': params - } - + + memberships = [_membership_to_dict(m) for m in webex_api.memberships.list(**params)] + return create_success_response( + data={'memberships': memberships}, + metadata={'count': len(memberships), 'filters_applied': params} + ) except Exception as e: - return { - 'success': False, - 'error': str(e), - 'memberships': [], - 'count': 0 - } + return _map_exception_to_error(e) def add_webex_membership( @@ -83,68 +93,39 @@ def add_webex_membership( ) -> Dict[str, Any]: """ Add a person to a Webex room. - + Args: room_id: Room ID to add person to (required) person_id: Person ID to add (use this OR person_email) person_email: Person email to add (use this OR person_id) is_moderator: Whether to make the person a moderator (optional) - + Returns: - Dictionary containing the created membership details + Standardized response dictionary with success/error information """ try: - # Validate required parameters if not (person_id or person_email): - return { - 'success': False, - 'error': 'Must specify either person_id or person_email', - 'membership': None - } - - # Build parameters dict - params = {'roomId': room_id} - + return create_error_response( + error_code=WebexErrorCodes.INVALID_ARGUMENTS, + message="Must specify either person_id or person_email", + details={"required_one_of": ["person_id", "person_email"]} + ) + + params: Dict[str, Any] = {'roomId': room_id} if person_id: params['personId'] = person_id elif person_email: params['personEmail'] = person_email - if is_moderator is not None: params['isModerator'] = is_moderator - - # Call the Webex API - membership_response = webex_api.memberships.create(**params) - - # Convert the response to a dictionary - membership_dict = { - 'id': membership_response.id, - 'roomId': membership_response.roomId, - 'personId': membership_response.personId, - 'personEmail': membership_response.personEmail, - 'personDisplayName': membership_response.personDisplayName, - 'personOrgId': membership_response.personOrgId, - 'isModerator': membership_response.isModerator, - 'isMonitor': membership_response.isMonitor, - 'created': membership_response.created - } - - # Add optional fields if they exist - if hasattr(membership_response, 'roomType') and membership_response.roomType: - membership_dict['roomType'] = membership_response.roomType - - return { - 'success': True, - 'membership': membership_dict, - 'parameters_used': params - } - + + membership = webex_api.memberships.create(**params) + return create_success_response( + data={'membership': _membership_to_dict(membership)}, + metadata={'operation': 'add_membership', 'parameters_used': params} + ) except Exception as e: - return { - 'success': False, - 'error': str(e), - 'membership': None - } + return _map_exception_to_error(e) def update_webex_membership( @@ -154,68 +135,65 @@ def update_webex_membership( ) -> Dict[str, Any]: """ Update an existing Webex membership (change moderator/monitor status). - + Args: membership_id: Membership ID to update (required) is_moderator: Whether the person should be a moderator (optional) is_monitor: Whether the person should be a monitor (optional) - + Returns: - Dictionary containing the updated membership details + Standardized response dictionary with success/error information """ try: - # Build parameters dict - only include fields that are being updated - params = {} - + params: Dict[str, Any] = {} if is_moderator is not None: params['isModerator'] = is_moderator if is_monitor is not None: params['isMonitor'] = is_monitor - - # Validate that at least one field is being updated + if not params: - return { - 'success': False, - 'error': 'Must specify at least one field to update (is_moderator or is_monitor)', - 'membership': None - } - - # Call the Webex API - membership_response = webex_api.memberships.update(membershipId=membership_id, **params) - - # Convert the response to a dictionary - membership_dict = { - 'id': membership_response.id, - 'roomId': membership_response.roomId, - 'personId': membership_response.personId, - 'personEmail': membership_response.personEmail, - 'personDisplayName': membership_response.personDisplayName, - 'personOrgId': membership_response.personOrgId, - 'isModerator': membership_response.isModerator, - 'isMonitor': membership_response.isMonitor, - 'created': membership_response.created - } - - # Add optional fields if they exist - if hasattr(membership_response, 'roomType') and membership_response.roomType: - membership_dict['roomType'] = membership_response.roomType - - return { - 'success': True, - 'membership': membership_dict, - 'membership_id': membership_id, - 'parameters_used': params - } - + return create_error_response( + error_code=WebexErrorCodes.INVALID_ARGUMENTS, + message="Must specify at least one field to update (is_moderator or is_monitor)" + ) + + membership = webex_api.memberships.update(membershipId=membership_id, **params) + return create_success_response( + data={'membership': _membership_to_dict(membership)}, + metadata={'operation': 'update_membership', 'membership_id': membership_id, + 'parameters_used': params} + ) except Exception as e: - return { - 'success': False, - 'error': str(e), - 'membership': None - } + return _map_exception_to_error(e) -# Space membership aliases - these provide the same functionality but with "space" terminology +def delete_webex_membership(membership_id: str) -> Dict[str, Any]: + """ + Remove a person from a Webex room by deleting their membership. + + Args: + membership_id: Membership ID to delete (required). Use list_webex_memberships + to find the membership ID for a person in a room. + + Returns: + Standardized response dictionary with success/error information + """ + try: + if not membership_id: + return create_error_response( + error_code=WebexErrorCodes.INVALID_ARGUMENTS, + message="membership_id is required" + ) + webex_api.memberships.delete(membershipId=membership_id) + return create_success_response( + data={'deleted': True, 'membership_id': membership_id}, + metadata={'operation': 'delete_membership'} + ) + except Exception as e: + return _map_exception_to_error(e) + + +# Space membership aliases — "room" and "space" are synonymous in Webex def list_webex_space_memberships( space_id: Optional[str] = None, @@ -225,23 +203,20 @@ def list_webex_space_memberships( ) -> Dict[str, Any]: """ List memberships for a space or person. - Note: This is an alias for list_webex_memberships - "room" and "space" are synonymous in Webex. - + Note: This is an alias for list_webex_memberships — "room" and "space" are synonymous in Webex. + Args: space_id: Space ID to get memberships for person_id: Person ID to get memberships for person_email: Person email to get memberships for max_results: Maximum number of memberships to return (default 100, max 1000) - + Returns: - Dictionary containing the list of memberships and metadata + Standardized response dictionary with success/error information """ - # Call the underlying memberships function with mapped parameters return list_webex_memberships( - room_id=space_id, # Map space_id to room_id - person_id=person_id, - person_email=person_email, - max_results=max_results + room_id=space_id, person_id=person_id, + person_email=person_email, max_results=max_results ) @@ -253,21 +228,18 @@ def add_webex_space_membership( ) -> Dict[str, Any]: """ Add a person to a Webex space. - Note: This is an alias for add_webex_membership - "room" and "space" are synonymous in Webex. - + Note: This is an alias for add_webex_membership — "room" and "space" are synonymous in Webex. + Args: space_id: Space ID to add person to (required) person_id: Person ID to add (use this OR person_email) person_email: Person email to add (use this OR person_id) is_moderator: Whether to make the person a moderator (optional) - + Returns: - Dictionary containing the created membership details + Standardized response dictionary with success/error information """ - # Call the underlying membership function with mapped parameters return add_webex_membership( - room_id=space_id, # Map space_id to room_id - person_id=person_id, - person_email=person_email, - is_moderator=is_moderator + room_id=space_id, person_id=person_id, + person_email=person_email, is_moderator=is_moderator ) diff --git a/tools/messages.py b/tools/messages.py index 1b5f536..5596dc1 100644 --- a/tools/messages.py +++ b/tools/messages.py @@ -6,100 +6,163 @@ def format_mention_by_email(email: str, display_name: Optional[str] = None) -> str: - """ - Format a mention by email address for Webex messages. - - Args: - email: Email address of the person to mention - display_name: Optional display name to show in the mention - - Returns: - Properly formatted mention string - """ if display_name: return f"<@personEmail:{email}|{display_name}>" - else: - return f"<@personEmail:{email}>" + return f"<@personEmail:{email}>" def format_mention_by_person_id(person_id: str, display_name: Optional[str] = None) -> str: - """ - Format a mention by person ID for Webex messages. - - Args: - person_id: Person ID to mention - display_name: Optional display name to show in the mention - - Returns: - Properly formatted mention string - """ if display_name: return f"<@personId:{person_id}|{display_name}>" - else: - return f"<@personId:{person_id}>" + return f"<@personId:{person_id}>" def format_mention_all() -> str: - """ - Format a mention for all people in a room. - - Returns: - Properly formatted @all mention string - """ return "<@all>" +def _build_mention_strings(mentions: List[Dict[str, str]]) -> List[str]: + """Convert a mentions list into formatted mention strings.""" + result = [] + for mention in mentions: + mention_type = mention.get('type', '').lower() + display_name = mention.get('display_name') + if mention_type == 'email': + email = mention.get('value', '') + if email: + result.append(format_mention_by_email(email, display_name)) + elif mention_type == 'person_id': + person_id = mention.get('value', '') + if person_id: + result.append(format_mention_by_person_id(person_id, display_name)) + elif mention_type == 'all': + result.append(format_mention_all()) + return result + + def create_message_with_mentions( base_message: str, mentions: Optional[List[Dict[str, str]]] = None ) -> str: """ - Create a message with properly formatted mentions. - + Create a message with properly formatted mentions prepended. + Args: base_message: The base message text - mentions: List of mention dictionaries with keys: + mentions: List of mention dicts with keys: - type: "email", "person_id", or "all" - value: email address or person ID (not needed for "all") - display_name: optional display name - + Returns: Message with properly formatted mentions """ if not mentions: return base_message - - message_parts = [base_message] - - for mention in mentions: - mention_type = mention.get('type', '').lower() - - if mention_type == 'email': - email = mention.get('value', '') - display_name = mention.get('display_name') - if email: - formatted_mention = format_mention_by_email(email, display_name) - message_parts.append(formatted_mention) - - elif mention_type == 'person_id': - person_id = mention.get('value', '') - display_name = mention.get('display_name') - if person_id: - formatted_mention = format_mention_by_person_id(person_id, display_name) - message_parts.append(formatted_mention) - - elif mention_type == 'all': - formatted_mention = format_mention_all() - message_parts.append(formatted_mention) - - return ' '.join(message_parts) + strings = _build_mention_strings(mentions) + if strings: + return ' '.join(strings) + ' ' + base_message + return base_message + + +def _map_exception_to_error(e: Exception) -> Dict[str, Any]: + """Map a caught exception to a structured error response.""" + error_str = str(e).lower() + if 'rate limit' in error_str or 'too many requests' in error_str: + return create_error_response( + error_code=WebexErrorCodes.RATE_LIMITED, + message="API rate limit exceeded. Please retry after delay.", + temporary=True, + retry_after_seconds=60 + ) + if 'unauthorized' in error_str or 'invalid token' in error_str: + return create_error_response( + error_code=WebexErrorCodes.UNAUTHORIZED, + message="Invalid or expired bot token. Please check WEBEX_ACCESS_TOKEN." + ) + if 'not found' in error_str or '404' in error_str: + return create_error_response( + error_code=WebexErrorCodes.NOT_FOUND, + message="Room or person not found. Please verify the ID/email is correct." + ) + if 'forbidden' in error_str or '403' in error_str: + return create_error_response( + error_code=WebexErrorCodes.FORBIDDEN, + message="Bot lacks permission for this operation. Ensure bot is added to the room." + ) + if 'network' in error_str or 'connection' in error_str: + return create_error_response( + error_code=WebexErrorCodes.NETWORK_ERROR, + message="Network connectivity issue. Please retry.", + temporary=True, + retry_after_seconds=30 + ) + return create_error_response( + error_code=WebexErrorCodes.WEBEX_API_ERROR, + message=f"Webex API error: {e}", + temporary=True, + details={'original_error': str(e)} + ) + + +def _build_message_params( + room_id: Optional[str], + to_person_id: Optional[str], + to_person_email: Optional[str], + text: Optional[str], + markdown: Optional[str], + html: Optional[str], + files: Optional[str], + parent_id: Optional[str], +) -> Dict[str, Any]: + """Assemble the keyword-argument dict for webex_api.messages.create.""" + params: Dict[str, Any] = {} + if room_id: + params['roomId'] = room_id + elif to_person_id: + params['toPersonId'] = to_person_id + elif to_person_email: + params['toPersonEmail'] = to_person_email + if text: + params['text'] = text + if markdown: + params['markdown'] = markdown + if html: + params['html'] = html + if files: + params['files'] = [files] if isinstance(files, str) else files + if parent_id: + params['parentId'] = parent_id + return params + + +def _message_to_dict(message) -> Dict[str, Any]: + """Convert a Webex SDK message object to a plain dict.""" + d: Dict[str, Any] = { + 'id': message.id, + 'roomId': message.roomId, + 'text': message.text, + 'personId': message.personId, + 'personEmail': message.personEmail, + 'created': ( + message.created.isoformat() + if hasattr(message.created, 'isoformat') + else str(message.created) + ), + } + for attr in ('roomType', 'markdown', 'html', 'files', 'parentId', + 'mentionedPeople', 'mentionedGroups'): + val = getattr(message, attr, None) + if val: + d[attr] = val + return d def send_webex_message( room_id: Optional[str] = None, to_person_id: Optional[str] = None, to_person_email: Optional[str] = None, - text: Optional[str] = None, + text: Optional[str] = None, markdown: Optional[str] = None, html: Optional[str] = None, files: Optional[str] = None, @@ -107,7 +170,7 @@ def send_webex_message( ) -> Dict[str, Any]: """ Send a message to a Webex room or person. - + Args: room_id: Room ID to send the message to (use this OR to_person_id/to_person_email) to_person_id: Person ID to send a direct message to @@ -115,19 +178,19 @@ def send_webex_message( text: Plain text message content markdown: Markdown formatted message content html: HTML formatted message content (for buttons/cards) - files: URL or file path to attach (single file as string) + files: URL to attach (single file URL as a string) parent_id: Parent message ID for threaded replies - + Returns: Standardized response dictionary with success/error information - + Examples: # Send to room result = send_webex_message( room_id="Y2lzY29zcGFyazovL3VzL1JPT00vYmJjZWIx", text="Hello team!" ) - + # Send direct message result = send_webex_message( to_person_email="user@company.com", @@ -135,89 +198,42 @@ def send_webex_message( ) """ try: - # Validate required parameters - destination if not (room_id or to_person_id or to_person_email): return create_error_response( error_code=WebexErrorCodes.INVALID_ARGUMENTS, message="Must specify either room_id, to_person_id, or to_person_email", - details={ - "required_fields": ["room_id", "to_person_id", "to_person_email"], - "provided_fields": [] - } + details={"required_one_of": ["room_id", "to_person_id", "to_person_email"]} ) - - # Validate required parameters - content + if not (text or markdown or html or files): return create_error_response( error_code=WebexErrorCodes.MISSING_REQUIRED_FIELD, message="Must specify at least one of: text, markdown, html, or files", - details={ - "required_fields": ["text", "markdown", "html", "files"], - "provided_fields": [] - } + details={"required_one_of": ["text", "markdown", "html", "files"]} ) - - # Validate content length limits + if text and len(text) > 7439: return create_error_response( error_code=WebexErrorCodes.INVALID_FIELD_VALUE, message=f"Text content exceeds maximum length of 7439 characters (provided: {len(text)})", - details={ - "field": "text", - "max_length": 7439, - "provided_length": len(text) - } + details={"field": "text", "max_length": 7439, "provided_length": len(text)} ) - + if markdown and len(markdown) > 7439: return create_error_response( error_code=WebexErrorCodes.INVALID_FIELD_VALUE, message=f"Markdown content exceeds maximum length of 7439 characters (provided: {len(markdown)})", - details={ - "field": "markdown", - "max_length": 7439, - "provided_length": len(markdown) - } + details={"field": "markdown", "max_length": 7439, "provided_length": len(markdown)} ) - - # Build parameters dict, only including non-None values - params = {} - - # Destination parameters (mutually exclusive) - if room_id: - params['roomId'] = room_id - elif to_person_id: - params['toPersonId'] = to_person_id - elif to_person_email: - params['toPersonEmail'] = to_person_email - - # Content parameters - if text: - params['text'] = text - if markdown: - params['markdown'] = markdown - if html: - params['html'] = html - if files: - params['files'] = files - if parent_id: - params['parentId'] = parent_id - - # Send the message + + params = _build_message_params( + room_id, to_person_id, to_person_email, + text, markdown, html, files, parent_id + ) message = webex_api.messages.create(**params) - + return create_success_response( - data={ - 'id': message.id, - 'roomId': message.roomId, - 'text': message.text, - 'markdown': getattr(message, 'markdown', None), - 'html': getattr(message, 'html', None), - 'files': getattr(message, 'files', None), - 'personId': message.personId, - 'personEmail': message.personEmail, - 'created': message.created.isoformat() if hasattr(message.created, 'isoformat') else str(message.created) - }, + data=_message_to_dict(message), metadata={ 'operation': 'send_message', 'destination_type': 'room' if room_id else 'direct', @@ -226,112 +242,16 @@ def send_webex_message( 'is_threaded': bool(parent_id) } ) - - except Exception as e: - error_str = str(e).lower() - - # Check for specific error types and return appropriate responses - if 'rate limit' in error_str or 'too many requests' in error_str: - return create_error_response( - error_code=WebexErrorCodes.RATE_LIMITED, - message="API rate limit exceeded. Please retry after delay.", - temporary=True, - retry_after_seconds=60 - ) - - elif 'unauthorized' in error_str or 'invalid token' in error_str: - return create_error_response( - error_code=WebexErrorCodes.UNAUTHORIZED, - message="Invalid or expired bot token. Please check WEBEX_ACCESS_TOKEN." - ) - - elif 'not found' in error_str or '404' in error_str: - return create_error_response( - error_code=WebexErrorCodes.NOT_FOUND, - message="Room or person not found. Please verify the ID/email is correct." - ) - - elif 'forbidden' in error_str or '403' in error_str: - return create_error_response( - error_code=WebexErrorCodes.FORBIDDEN, - message="Bot lacks permission for this operation. Ensure bot is added to the room." - ) - - elif 'network' in error_str or 'connection' in error_str: - return create_error_response( - error_code=WebexErrorCodes.NETWORK_ERROR, - message="Network connectivity issue. Please retry.", - temporary=True, - retry_after_seconds=30 - ) - - else: - return create_error_response( - error_code=WebexErrorCodes.WEBEX_API_ERROR, - message=f"Webex API error: {str(e)}", - temporary=True, - details={'original_error': str(e)} - ) - - # Content parameters - if text: - params['text'] = text - if markdown: - params['markdown'] = markdown - if html: - params['html'] = html - if files: - params['files'] = [files] if isinstance(files, str) else files - if parent_id: - params['parentId'] = parent_id - - # Call the Webex API - message_response = webex_api.messages.create(**params) - - # Convert the response to a dictionary - message_dict = { - 'id': message_response.id, - 'roomId': message_response.roomId, - 'roomType': message_response.roomType, - 'text': message_response.text, - 'personId': message_response.personId, - 'personEmail': message_response.personEmail, - 'created': message_response.created - } - - # Add optional fields if they exist - if hasattr(message_response, 'markdown') and message_response.markdown: - message_dict['markdown'] = message_response.markdown - if hasattr(message_response, 'html') and message_response.html: - message_dict['html'] = message_response.html - if hasattr(message_response, 'files') and message_response.files: - message_dict['files'] = message_response.files - if hasattr(message_response, 'parentId') and message_response.parentId: - message_dict['parentId'] = message_response.parentId - if hasattr(message_response, 'mentionedPeople') and message_response.mentionedPeople: - message_dict['mentionedPeople'] = message_response.mentionedPeople - if hasattr(message_response, 'mentionedGroups') and message_response.mentionedGroups: - message_dict['mentionedGroups'] = message_response.mentionedGroups - - return { - 'success': True, - 'message': message_dict, - 'parameters_used': params - } - + except Exception as e: - return { - 'success': False, - 'error': str(e), - 'message': None - } + return _map_exception_to_error(e) def send_webex_message_with_mentions( room_id: Optional[str] = None, to_person_id: Optional[str] = None, to_person_email: Optional[str] = None, - text: Optional[str] = None, + text: Optional[str] = None, markdown: Optional[str] = None, html: Optional[str] = None, files: Optional[str] = None, @@ -340,7 +260,7 @@ def send_webex_message_with_mentions( ) -> Dict[str, Any]: """ Send a message to a Webex room or person with proper mention support. - + Args: room_id: Room ID to send the message to (use this OR to_person_id/to_person_email) to_person_id: Person ID to send a direct message to @@ -348,154 +268,58 @@ def send_webex_message_with_mentions( text: Plain text message content markdown: Markdown formatted message content html: HTML formatted message content (for buttons/cards) - files: URL or file path to attach (single file as string) + files: URL to attach (single file URL as a string) parent_id: Parent message ID for threaded replies - mentions: List of mention dictionaries with keys: + mentions: List of mention dicts with keys: - type: "email", "person_id", or "all" - value: email address or person ID (not needed for "all") - display_name: optional display name - + Returns: - Dictionary containing the sent message details and metadata + Standardized response dictionary with success/error information """ try: - # Validate required parameters if not (room_id or to_person_id or to_person_email): - return { - 'success': False, - 'error': 'Must specify either room_id, to_person_id, or to_person_email', - 'message': None - } - + return create_error_response( + error_code=WebexErrorCodes.INVALID_ARGUMENTS, + message="Must specify either room_id, to_person_id, or to_person_email", + details={"required_one_of": ["room_id", "to_person_id", "to_person_email"]} + ) + if not (text or markdown or html or files): - return { - 'success': False, - 'error': 'Must specify at least one of: text, markdown, html, or files', - 'message': None - } - - # Build parameters dict, only including non-None values - params = {} - - # Destination parameters (mutually exclusive) - if room_id: - params['roomId'] = room_id - elif to_person_id: - params['toPersonId'] = to_person_id - elif to_person_email: - params['toPersonEmail'] = to_person_email - - # Process mentions and update content accordingly + return create_error_response( + error_code=WebexErrorCodes.MISSING_REQUIRED_FIELD, + message="Must specify at least one of: text, markdown, html, or files", + details={"required_one_of": ["text", "markdown", "html", "files"]} + ) + + # Apply mentions to the primary content field if mentions: if markdown: - # Add mentions to markdown content - mention_strings = [] - for mention in mentions: - mention_type = mention.get('type', '').lower() - - if mention_type == 'email': - email = mention.get('value', '') - display_name = mention.get('display_name') - if email: - mention_strings.append(format_mention_by_email(email, display_name)) - - elif mention_type == 'person_id': - person_id = mention.get('value', '') - display_name = mention.get('display_name') - if person_id: - mention_strings.append(format_mention_by_person_id(person_id, display_name)) - - elif mention_type == 'all': - mention_strings.append(format_mention_all()) - - # Prepend mentions to the markdown content - if mention_strings: - params['markdown'] = ' '.join(mention_strings) + ' ' + markdown - else: - params['markdown'] = markdown - + markdown = create_message_with_mentions(markdown, mentions) elif text: - # Add mentions to text content - mention_strings = [] - for mention in mentions: - mention_type = mention.get('type', '').lower() - - if mention_type == 'email': - email = mention.get('value', '') - display_name = mention.get('display_name') - if email: - mention_strings.append(format_mention_by_email(email, display_name)) - - elif mention_type == 'person_id': - person_id = mention.get('value', '') - display_name = mention.get('display_name') - if person_id: - mention_strings.append(format_mention_by_person_id(person_id, display_name)) - - elif mention_type == 'all': - mention_strings.append(format_mention_all()) - - # Prepend mentions to the text content - if mention_strings: - params['text'] = ' '.join(mention_strings) + ' ' + text - else: - params['text'] = text - else: - # No mentions, use original content - if text: - params['text'] = text - if markdown: - params['markdown'] = markdown - - # Add other content parameters - if html: - params['html'] = html - if files: - params['files'] = [files] if isinstance(files, str) else files - if parent_id: - params['parentId'] = parent_id - - # Call the Webex API - message_response = webex_api.messages.create(**params) - - # Convert the response to a dictionary - message_dict = { - 'id': message_response.id, - 'roomId': message_response.roomId, - 'roomType': message_response.roomType, - 'text': message_response.text, - 'personId': message_response.personId, - 'personEmail': message_response.personEmail, - 'created': message_response.created - } - - # Add optional fields if they exist - if hasattr(message_response, 'markdown') and message_response.markdown: - message_dict['markdown'] = message_response.markdown - if hasattr(message_response, 'html') and message_response.html: - message_dict['html'] = message_response.html - if hasattr(message_response, 'files') and message_response.files: - message_dict['files'] = message_response.files - if hasattr(message_response, 'parentId') and message_response.parentId: - message_dict['parentId'] = message_response.parentId - if hasattr(message_response, 'mentionedPeople') and message_response.mentionedPeople: - message_dict['mentionedPeople'] = message_response.mentionedPeople - if hasattr(message_response, 'mentionedGroups') and message_response.mentionedGroups: - message_dict['mentionedGroups'] = message_response.mentionedGroups - - return { - 'success': True, - 'message': message_dict, - 'parameters_used': params, - 'mentions_processed': mentions or [] - } - + text = create_message_with_mentions(text, mentions) + + params = _build_message_params( + room_id, to_person_id, to_person_email, + text, markdown, html, files, parent_id + ) + message = webex_api.messages.create(**params) + + return create_success_response( + data=_message_to_dict(message), + metadata={ + 'operation': 'send_message_with_mentions', + 'destination_type': 'room' if room_id else 'direct', + 'content_type': 'markdown' if markdown else 'html' if html else 'text', + 'has_files': bool(files), + 'is_threaded': bool(parent_id), + 'mentions_processed': mentions or [] + } + ) + except Exception as e: - return { - 'success': False, - 'error': str(e), - 'message': None - } + return _map_exception_to_error(e) def list_webex_messages( @@ -507,21 +331,19 @@ def list_webex_messages( ) -> Dict[str, Any]: """ List messages from a Webex room. - + Args: room_id: Room ID to get messages from (required) mentioned_people: Person ID to filter messages that mention them before: Get messages before this date (ISO 8601 format) before_message: Get messages before this message ID max_results: Maximum number of messages to return (default 50, max 1000) - + Returns: - Dictionary containing the list of messages and metadata + Standardized response dictionary with success/error information """ try: - # Build parameters dict - params = {'roomId': room_id} - + params: Dict[str, Any] = {'roomId': room_id} if mentioned_people: params['mentionedPeople'] = mentioned_people if before: @@ -530,57 +352,45 @@ def list_webex_messages( params['beforeMessage'] = before_message if max_results: params['max'] = max_results - - # Call the Webex API + messages_response = webex_api.messages.list(**params) - - # Convert the response to a list of dictionaries - messages_list = [] - for message in messages_response: - message_dict = { - 'id': message.id, - 'roomId': message.roomId, - 'roomType': message.roomType, - 'text': message.text, - 'personId': message.personId, - 'personEmail': message.personEmail, - 'created': message.created - } - - # Add optional fields if they exist - if hasattr(message, 'markdown') and message.markdown: - message_dict['markdown'] = message.markdown - if hasattr(message, 'html') and message.html: - message_dict['html'] = message.html - if hasattr(message, 'files') and message.files: - message_dict['files'] = message.files - if hasattr(message, 'parentId') and message.parentId: - message_dict['parentId'] = message.parentId - if hasattr(message, 'mentionedPeople') and message.mentionedPeople: - message_dict['mentionedPeople'] = message.mentionedPeople - if hasattr(message, 'mentionedGroups') and message.mentionedGroups: - message_dict['mentionedGroups'] = message.mentionedGroups - - messages_list.append(message_dict) - - return { - 'success': True, - 'messages': messages_list, - 'count': len(messages_list), - 'room_id': room_id, - 'filters_applied': params - } - + messages_list = [_message_to_dict(m) for m in messages_response] + + return create_success_response( + data={'messages': messages_list, 'room_id': room_id}, + metadata={'count': len(messages_list), 'filters_applied': params} + ) + except Exception as e: - return { - 'success': False, - 'error': str(e), - 'messages': [], - 'count': 0 - } + return _map_exception_to_error(e) + +def delete_webex_message(message_id: str) -> Dict[str, Any]: + """ + Delete a Webex message. -# Space message aliases - these provide the same functionality but with "space" terminology + Args: + message_id: ID of the message to delete (required) + + Returns: + Standardized response dictionary with success/error information + """ + try: + if not message_id: + return create_error_response( + error_code=WebexErrorCodes.INVALID_ARGUMENTS, + message="message_id is required" + ) + webex_api.messages.delete(messageId=message_id) + return create_success_response( + data={'deleted': True, 'message_id': message_id}, + metadata={'operation': 'delete_message'} + ) + except Exception as e: + return _map_exception_to_error(e) + + +# Space message aliases — "room" and "space" are synonymous in Webex def send_webex_space_message( space_id: Optional[str] = None, @@ -594,8 +404,8 @@ def send_webex_space_message( ) -> Dict[str, Any]: """ Send a message to a Webex space or person. - Note: This is an alias for send_webex_message - "room" and "space" are synonymous in Webex. - + Note: This is an alias for send_webex_message — "room" and "space" are synonymous in Webex. + Args: space_id: Space ID to send the message to (use this OR to_person_id/to_person_email) to_person_id: Person ID to send a direct message to @@ -603,15 +413,14 @@ def send_webex_space_message( text: Plain text message content markdown: Markdown formatted message content html: HTML formatted message content (for buttons/cards) - files: URL or file path to attach (single file as string) + files: URL to attach (single file URL as a string) parent_id: Parent message ID for threaded replies - + Returns: - Dictionary containing the sent message details and metadata + Standardized response dictionary with success/error information """ - # Call the underlying message function with mapped parameters return send_webex_message( - room_id=space_id, # Map space_id to room_id + room_id=space_id, to_person_id=to_person_id, to_person_email=to_person_email, text=text, @@ -631,29 +440,22 @@ def list_webex_space_messages( ) -> Dict[str, Any]: """ List messages from a Webex space. - Note: This is an alias for list_webex_messages - "room" and "space" are synonymous in Webex. - + Note: This is an alias for list_webex_messages — "room" and "space" are synonymous in Webex. + Args: space_id: Space ID to get messages from (required) mentioned_people: Person ID to filter messages that mention them before: Get messages before this date (ISO 8601 format) before_message: Get messages before this message ID max_results: Maximum number of messages to return (default 50, max 1000) - + Returns: - Dictionary containing the list of messages and metadata + Standardized response dictionary with success/error information """ - # Call the underlying messages function with mapped parameters - result = list_webex_messages( - room_id=space_id, # Map space_id to room_id + return list_webex_messages( + room_id=space_id, mentioned_people=mentioned_people, before=before, before_message=before_message, max_results=max_results ) - - # Update the response to use "space" terminology - if result.get('success'): - result['space_id'] = result.pop('room_id', None) - - return result diff --git a/tools/people.py b/tools/people.py index 1475aa0..79be993 100644 --- a/tools/people.py +++ b/tools/people.py @@ -2,71 +2,73 @@ Webex People management tools. """ from typing import Optional, Dict, Any -from .common import webex_api +from .common import webex_api, create_error_response, create_success_response, WebexErrorCodes + + +def _map_exception_to_error(e: Exception) -> Dict[str, Any]: + error_str = str(e).lower() + if 'unauthorized' in error_str or 'invalid token' in error_str: + return create_error_response(WebexErrorCodes.UNAUTHORIZED, + "Invalid or expired bot token.") + if 'not found' in error_str or '404' in error_str: + return create_error_response(WebexErrorCodes.NOT_FOUND, + "Person not found. Verify the ID or email is correct.") + if 'forbidden' in error_str or '403' in error_str: + return create_error_response(WebexErrorCodes.FORBIDDEN, + "Bot lacks permission for this operation.") + if 'rate limit' in error_str or 'too many requests' in error_str: + return create_error_response(WebexErrorCodes.RATE_LIMITED, + "API rate limit exceeded. Please retry after delay.", + temporary=True, retry_after_seconds=60) + if 'network' in error_str or 'connection' in error_str: + return create_error_response(WebexErrorCodes.NETWORK_ERROR, + "Network connectivity issue. Please retry.", + temporary=True, retry_after_seconds=30) + return create_error_response(WebexErrorCodes.WEBEX_API_ERROR, + f"Webex API error: {e}", + temporary=True, details={'original_error': str(e)}) + + +def _person_to_dict(person) -> Dict[str, Any]: + d: Dict[str, Any] = { + 'id': person.id, + 'emails': person.emails, + 'displayName': person.displayName, + 'nickName': person.nickName, + 'firstName': person.firstName, + 'lastName': person.lastName, + 'avatar': person.avatar, + 'orgId': person.orgId, + 'created': person.created, + 'status': person.status, + 'type': person.type, + } + for attr in ('userName', 'lastModified', 'roles', 'licenses', + 'phoneNumbers', 'extension', 'locationId', 'addresses', 'timezone'): + val = getattr(person, attr, None) + if val: + d[attr] = val + return d def get_webex_me() -> Dict[str, Any]: """ - Get information about the authenticated Webex user. - + Get information about the authenticated Webex bot. + Returns: - Dictionary containing the authenticated user's information + Standardized response dictionary with success/error information """ try: - # Call the Webex API to get user info - me_response = webex_api.people.me() - - # Convert the response to a dictionary - me_dict = { - 'id': me_response.id, - 'emails': me_response.emails, - 'displayName': me_response.displayName, - 'nickName': me_response.nickName, - 'firstName': me_response.firstName, - 'lastName': me_response.lastName, - 'avatar': me_response.avatar, - 'orgId': me_response.orgId, - 'created': me_response.created, - 'lastModified': me_response.lastModified, - 'status': me_response.status, - 'type': me_response.type - } - - # Add optional fields if they exist - if hasattr(me_response, 'userName') and me_response.userName: - me_dict['userName'] = me_response.userName - if hasattr(me_response, 'roles') and me_response.roles: - me_dict['roles'] = me_response.roles - if hasattr(me_response, 'licenses') and me_response.licenses: - me_dict['licenses'] = me_response.licenses - if hasattr(me_response, 'phoneNumbers') and me_response.phoneNumbers: - me_dict['phoneNumbers'] = me_response.phoneNumbers - if hasattr(me_response, 'extension') and me_response.extension: - me_dict['extension'] = me_response.extension - if hasattr(me_response, 'locationId') and me_response.locationId: - me_dict['locationId'] = me_response.locationId - if hasattr(me_response, 'addresses') and me_response.addresses: - me_dict['addresses'] = me_response.addresses - if hasattr(me_response, 'timezone') and me_response.timezone: - me_dict['timezone'] = me_response.timezone - - return { - 'success': True, - 'user': me_dict - } - + me = webex_api.people.me() + return create_success_response(data={'user': _person_to_dict(me)}) except Exception as e: - return { - 'success': False, - 'error': str(e), - 'user': None - } + return _map_exception_to_error(e) def list_webex_people( email: Optional[str] = None, display_name: Optional[str] = None, - id: Optional[str] = None, + person_id: Optional[str] = None, org_id: Optional[str] = None, calling_data: Optional[bool] = None, location_id: Optional[str] = None, @@ -74,28 +76,27 @@ def list_webex_people( ) -> Dict[str, Any]: """ List people in the organization or search for specific people. - + Args: email: Email address to search for display_name: Display name to search for - id: Person ID to get specific person + person_id: Person ID to look up a specific person org_id: Organization ID to filter by calling_data: Include calling data in response location_id: Location ID to filter by max_results: Maximum number of people to return (default 100, max 1000) - + Returns: - Dictionary containing the list of people and metadata + Standardized response dictionary with success/error information """ try: - # Build parameters dict - params = {} + params: Dict[str, Any] = {} if email: params['email'] = email if display_name: params['displayName'] = display_name - if id: - params['id'] = id + if person_id: + params['id'] = person_id if org_id: params['orgId'] = org_id if calling_data is not None: @@ -104,60 +105,11 @@ def list_webex_people( params['locationId'] = location_id if max_results: params['max'] = max_results - - # Call the Webex API - people_response = webex_api.people.list(**params) - - # Convert the response to a list of dictionaries - people_list = [] - for person in people_response: - person_dict = { - 'id': person.id, - 'emails': person.emails, - 'displayName': person.displayName, - 'nickName': person.nickName, - 'firstName': person.firstName, - 'lastName': person.lastName, - 'avatar': person.avatar, - 'orgId': person.orgId, - 'created': person.created, - 'status': person.status, - 'type': person.type - } - - # Add optional fields if they exist - if hasattr(person, 'userName') and person.userName: - person_dict['userName'] = person.userName - if hasattr(person, 'lastModified') and person.lastModified: - person_dict['lastModified'] = person.lastModified - if hasattr(person, 'roles') and person.roles: - person_dict['roles'] = person.roles - if hasattr(person, 'licenses') and person.licenses: - person_dict['licenses'] = person.licenses - if hasattr(person, 'phoneNumbers') and person.phoneNumbers: - person_dict['phoneNumbers'] = person.phoneNumbers - if hasattr(person, 'extension') and person.extension: - person_dict['extension'] = person.extension - if hasattr(person, 'locationId') and person.locationId: - person_dict['locationId'] = person.locationId - if hasattr(person, 'addresses') and person.addresses: - person_dict['addresses'] = person.addresses - if hasattr(person, 'timezone') and person.timezone: - person_dict['timezone'] = person.timezone - - people_list.append(person_dict) - - return { - 'success': True, - 'people': people_list, - 'count': len(people_list), - 'filters_applied': params - } - + + people = [_person_to_dict(p) for p in webex_api.people.list(**params)] + return create_success_response( + data={'people': people}, + metadata={'count': len(people), 'filters_applied': params} + ) except Exception as e: - return { - 'success': False, - 'error': str(e), - 'people': [], - 'count': 0 - } + return _map_exception_to_error(e) diff --git a/tools/rooms.py b/tools/rooms.py index bc94e91..00034c4 100644 --- a/tools/rooms.py +++ b/tools/rooms.py @@ -2,7 +2,49 @@ Webex Room/Space management tools. """ from typing import Optional, Dict, Any -from .common import webex_api +from .common import webex_api, create_error_response, create_success_response, WebexErrorCodes + + +def _map_exception_to_error(e: Exception) -> Dict[str, Any]: + error_str = str(e).lower() + if 'unauthorized' in error_str or 'invalid token' in error_str: + return create_error_response(WebexErrorCodes.UNAUTHORIZED, + "Invalid or expired bot token.") + if 'not found' in error_str or '404' in error_str: + return create_error_response(WebexErrorCodes.NOT_FOUND, + "Room not found. Verify the room ID is correct.") + if 'forbidden' in error_str or '403' in error_str: + return create_error_response(WebexErrorCodes.FORBIDDEN, + "Bot lacks permission for this operation.") + if 'rate limit' in error_str or 'too many requests' in error_str: + return create_error_response(WebexErrorCodes.RATE_LIMITED, + "API rate limit exceeded. Please retry after delay.", + temporary=True, retry_after_seconds=60) + if 'network' in error_str or 'connection' in error_str: + return create_error_response(WebexErrorCodes.NETWORK_ERROR, + "Network connectivity issue. Please retry.", + temporary=True, retry_after_seconds=30) + return create_error_response(WebexErrorCodes.WEBEX_API_ERROR, + f"Webex API error: {e}", + temporary=True, details={'original_error': str(e)}) + + +def _room_to_dict(room) -> Dict[str, Any]: + d: Dict[str, Any] = { + 'id': room.id, + 'title': room.title, + 'type': room.type, + 'isLocked': room.isLocked, + 'lastActivity': room.lastActivity, + 'created': room.created, + 'creatorId': room.creatorId, + } + for attr in ('teamId', 'sipAddress', 'description', 'isPublic', + 'isAnnouncementOnly', 'ownerId', 'classificationId'): + val = getattr(room, attr, None) + if val is not None: + d[attr] = val + return d def list_webex_rooms( @@ -12,20 +54,19 @@ def list_webex_rooms( max_results: Optional[int] = None ) -> Dict[str, Any]: """ - List Webex rooms that the authenticated user belongs to. - + List Webex rooms that the authenticated bot belongs to. + Args: team_id: Optional team ID to filter rooms by team - room_type: Optional room type filter ('direct' or 'group') + room_type: Optional room type filter ('direct' or 'group') sort_by: Optional sort order ('id', 'lastactivity', 'created') max_results: Optional maximum number of rooms to return (default 100, max 1000) - + Returns: - Dictionary containing the list of rooms and metadata + Standardized response dictionary with success/error information """ try: - # Build parameters dict, only including non-None values - params = {} + params: Dict[str, Any] = {} if team_id: params['teamId'] = team_id if room_type: @@ -34,47 +75,14 @@ def list_webex_rooms( params['sortBy'] = sort_by if max_results: params['max'] = max_results - - # Call the Webex API - rooms_response = webex_api.rooms.list(**params) - - # Convert the response to a list of dictionaries - rooms_list = [] - for room in rooms_response: - room_dict = { - 'id': room.id, - 'title': room.title, - 'type': room.type, - 'isLocked': room.isLocked, - 'lastActivity': room.lastActivity, - 'created': room.created, - 'creatorId': room.creatorId - } - - # Add optional fields if they exist - if hasattr(room, 'teamId') and room.teamId: - room_dict['teamId'] = room.teamId - if hasattr(room, 'sipAddress') and room.sipAddress: - room_dict['sipAddress'] = room.sipAddress - if hasattr(room, 'description') and room.description: - room_dict['description'] = room.description - - rooms_list.append(room_dict) - - return { - 'success': True, - 'rooms': rooms_list, - 'count': len(rooms_list), - 'filters_applied': params - } - + + rooms_list = [_room_to_dict(r) for r in webex_api.rooms.list(**params)] + return create_success_response( + data={'rooms': rooms_list}, + metadata={'count': len(rooms_list), 'filters_applied': params} + ) except Exception as e: - return { - 'success': False, - 'error': str(e), - 'rooms': [], - 'count': 0 - } + return _map_exception_to_error(e) def create_webex_room( @@ -89,35 +97,30 @@ def create_webex_room( ) -> Dict[str, Any]: """ Create a new Webex room. - + Args: title: Title of the room (required) team_id: Team ID to create room in (optional) classification_id: Classification for the room (optional) - is_locked: Whether the room is locked (optional, same as is_moderated) - is_moderated: Whether the room is moderated (optional, same as is_locked) + is_locked: Whether the room is locked (optional; same property as is_moderated) + is_moderated: Whether the room is moderated (optional; same property as is_locked) is_public: Whether the room is public (optional) - is_announcement_only: Whether the room is announcement-only (only moderators can post) (optional) + is_announcement_only: Whether only moderators can post (optional) description: Description of the room (optional) - + Returns: - Dictionary containing the created room details + Standardized response dictionary with success/error information """ try: - # Handle the is_moderated alias - both is_locked and is_moderated refer to the same Webex property if is_moderated is not None and is_locked is not None: - return { - 'success': False, - 'error': 'Cannot specify both is_locked and is_moderated - they are the same property. Use either one.', - 'room': None - } - - # Use is_moderated if provided, otherwise use is_locked + return create_error_response( + error_code=WebexErrorCodes.INVALID_ARGUMENTS, + message="Cannot specify both is_locked and is_moderated — they are the same property." + ) + final_is_locked = is_moderated if is_moderated is not None else is_locked - - # Build parameters dict - params = {'title': title} - + + params: Dict[str, Any] = {'title': title} if team_id: params['teamId'] = team_id if classification_id: @@ -130,45 +133,14 @@ def create_webex_room( params['isAnnouncementOnly'] = is_announcement_only if description: params['description'] = description - - # Call the Webex API - room_response = webex_api.rooms.create(**params) - - # Convert the response to a dictionary - room_dict = { - 'id': room_response.id, - 'title': room_response.title, - 'type': room_response.type, - 'isLocked': room_response.isLocked, - 'lastActivity': room_response.lastActivity, - 'created': room_response.created, - 'creatorId': room_response.creatorId - } - - # Add optional fields if they exist - if hasattr(room_response, 'teamId') and room_response.teamId: - room_dict['teamId'] = room_response.teamId - if hasattr(room_response, 'description') and room_response.description: - room_dict['description'] = room_response.description - if hasattr(room_response, 'sipAddress') and room_response.sipAddress: - room_dict['sipAddress'] = room_response.sipAddress - if hasattr(room_response, 'isPublic') and room_response.isPublic is not None: - room_dict['isPublic'] = room_response.isPublic - if hasattr(room_response, 'isAnnouncementOnly') and room_response.isAnnouncementOnly is not None: - room_dict['isAnnouncementOnly'] = room_response.isAnnouncementOnly - - return { - 'success': True, - 'room': room_dict, - 'parameters_used': params - } - + + room = webex_api.rooms.create(**params) + return create_success_response( + data={'room': _room_to_dict(room)}, + metadata={'operation': 'create_room', 'parameters_used': params} + ) except Exception as e: - return { - 'success': False, - 'error': str(e), - 'room': None - } + return _map_exception_to_error(e) def update_webex_room( @@ -183,35 +155,30 @@ def update_webex_room( ) -> Dict[str, Any]: """ Update an existing Webex room. - + Args: room_id: Room ID to update (required) title: New title for the room (optional) classification_id: New classification for the room (optional) - is_locked: Whether the room should be locked (optional, same as is_moderated) - is_moderated: Whether the room should be moderated (optional, same as is_locked) + is_locked: Whether the room should be locked (optional; same property as is_moderated) + is_moderated: Whether the room should be moderated (optional; same property as is_locked) is_public: Whether the room should be public (optional) - is_announcement_only: Whether the room should be announcement-only (only moderators can post) (optional) + is_announcement_only: Whether only moderators can post (optional) description: New description for the room (optional) - + Returns: - Dictionary containing the updated room details + Standardized response dictionary with success/error information """ try: - # Handle the is_moderated alias - both is_locked and is_moderated refer to the same Webex property if is_moderated is not None and is_locked is not None: - return { - 'success': False, - 'error': 'Cannot specify both is_locked and is_moderated - they are the same property. Use either one.', - 'room': None - } - - # Use is_moderated if provided, otherwise use is_locked + return create_error_response( + error_code=WebexErrorCodes.INVALID_ARGUMENTS, + message="Cannot specify both is_locked and is_moderated — they are the same property." + ) + final_is_locked = is_moderated if is_moderated is not None else is_locked - - # Build parameters dict - only include fields that are being updated - params = {} - + + params: Dict[str, Any] = {} if title is not None: params['title'] = title if classification_id is not None: @@ -224,115 +191,68 @@ def update_webex_room( params['isAnnouncementOnly'] = is_announcement_only if description is not None: params['description'] = description - - # Validate that at least one field is being updated + if not params: - return { - 'success': False, - 'error': 'Must specify at least one field to update (title, classification_id, is_locked/is_moderated, is_public, is_announcement_only, or description)', - 'room': None - } - - # Call the Webex API - room_response = webex_api.rooms.update(roomId=room_id, **params) - - # Convert the response to a dictionary - room_dict = { - 'id': room_response.id, - 'title': room_response.title, - 'type': room_response.type, - 'isLocked': room_response.isLocked, - 'lastActivity': room_response.lastActivity, - 'created': room_response.created, - 'creatorId': room_response.creatorId - } - - # Add optional fields if they exist - if hasattr(room_response, 'teamId') and room_response.teamId: - room_dict['teamId'] = room_response.teamId - if hasattr(room_response, 'description') and room_response.description: - room_dict['description'] = room_response.description - if hasattr(room_response, 'sipAddress') and room_response.sipAddress: - room_dict['sipAddress'] = room_response.sipAddress - if hasattr(room_response, 'isPublic') and room_response.isPublic is not None: - room_dict['isPublic'] = room_response.isPublic - if hasattr(room_response, 'isAnnouncementOnly') and room_response.isAnnouncementOnly is not None: - room_dict['isAnnouncementOnly'] = room_response.isAnnouncementOnly - - return { - 'success': True, - 'room': room_dict, - 'room_id': room_id, - 'parameters_used': params - } - + return create_error_response( + error_code=WebexErrorCodes.INVALID_ARGUMENTS, + message="Must specify at least one field to update." + ) + + room = webex_api.rooms.update(roomId=room_id, **params) + return create_success_response( + data={'room': _room_to_dict(room)}, + metadata={'operation': 'update_room', 'room_id': room_id, 'parameters_used': params} + ) except Exception as e: - return { - 'success': False, - 'error': str(e), - 'room': None - } + return _map_exception_to_error(e) -def get_webex_room( - room_id: str -) -> Dict[str, Any]: +def get_webex_room(room_id: str) -> Dict[str, Any]: """ Get detailed information about a specific Webex room. - + Args: room_id: Room ID to get details for (required) - + + Returns: + Standardized response dictionary with success/error information + """ + try: + room = webex_api.rooms.get(roomId=room_id) + return create_success_response( + data={'room': _room_to_dict(room)}, + metadata={'room_id': room_id} + ) + except Exception as e: + return _map_exception_to_error(e) + + +def delete_webex_room(room_id: str) -> Dict[str, Any]: + """ + Delete a Webex room. + + Args: + room_id: Room ID to delete (required) + Returns: - Dictionary containing the room details + Standardized response dictionary with success/error information """ try: - # Call the Webex API - room_response = webex_api.rooms.get(roomId=room_id) - - # Convert the response to a dictionary - room_dict = { - 'id': room_response.id, - 'title': room_response.title, - 'type': room_response.type, - 'isLocked': room_response.isLocked, - 'lastActivity': room_response.lastActivity, - 'created': room_response.created, - 'creatorId': room_response.creatorId - } - - # Add optional fields if they exist - if hasattr(room_response, 'teamId') and room_response.teamId: - room_dict['teamId'] = room_response.teamId - if hasattr(room_response, 'description') and room_response.description: - room_dict['description'] = room_response.description - if hasattr(room_response, 'sipAddress') and room_response.sipAddress: - room_dict['sipAddress'] = room_response.sipAddress - if hasattr(room_response, 'isPublic') and room_response.isPublic is not None: - room_dict['isPublic'] = room_response.isPublic - if hasattr(room_response, 'isAnnouncementOnly') and room_response.isAnnouncementOnly is not None: - room_dict['isAnnouncementOnly'] = room_response.isAnnouncementOnly - if hasattr(room_response, 'ownerId') and room_response.ownerId: - room_dict['ownerId'] = room_response.ownerId - if hasattr(room_response, 'classificationId') and room_response.classificationId: - room_dict['classificationId'] = room_response.classificationId - - return { - 'success': True, - 'room': room_dict, - 'room_id': room_id - } - + if not room_id: + return create_error_response( + error_code=WebexErrorCodes.INVALID_ARGUMENTS, + message="room_id is required" + ) + webex_api.rooms.delete(roomId=room_id) + return create_success_response( + data={'deleted': True, 'room_id': room_id}, + metadata={'operation': 'delete_room'} + ) except Exception as e: - return { - 'success': False, - 'error': str(e), - 'room': None - } + return _map_exception_to_error(e) -# Space aliases - these provide the same functionality as room tools but with "space" terminology -# This allows users to use either "room" or "space" interchangeably +# Space aliases — "room" and "space" are synonymous in Webex def list_webex_spaces( team_id: Optional[str] = None, @@ -341,30 +261,23 @@ def list_webex_spaces( max_results: Optional[int] = None ) -> Dict[str, Any]: """ - List Webex spaces that the authenticated user belongs to. - Note: This is an alias for list_webex_rooms - "room" and "space" are synonymous in Webex. - + List Webex spaces that the authenticated bot belongs to. + Note: This is an alias for list_webex_rooms — "room" and "space" are synonymous in Webex. + Args: team_id: Optional team ID to filter spaces by team - space_type: Optional space type filter ('direct' or 'group') + space_type: Optional space type filter ('direct' or 'group') sort_by: Optional sort order ('id', 'lastactivity', 'created') max_results: Optional maximum number of spaces to return (default 100, max 1000) - + Returns: - Dictionary containing the list of spaces and metadata + Standardized response dictionary with success/error information """ - # Call the underlying room function with mapped parameters result = list_webex_rooms( - team_id=team_id, - room_type=space_type, # Map space_type to room_type - sort_by=sort_by, - max_results=max_results + team_id=team_id, room_type=space_type, sort_by=sort_by, max_results=max_results ) - - # Update the response to use "space" terminology - if result.get('success'): - result['spaces'] = result.pop('rooms', []) - + if result.get('success') and 'data' in result: + result['data']['spaces'] = result['data'].pop('rooms', []) return result @@ -380,37 +293,28 @@ def create_webex_space( ) -> Dict[str, Any]: """ Create a new Webex space. - Note: This is an alias for create_webex_room - "room" and "space" are synonymous in Webex. - + Note: This is an alias for create_webex_room — "room" and "space" are synonymous in Webex. + Args: title: Title of the space (required) team_id: Team ID to create space in (optional) classification_id: Classification for the space (optional) - is_locked: Whether the space is locked (optional, same as is_moderated) - is_moderated: Whether the space is moderated (optional, same as is_locked) + is_locked: Whether the space is locked (optional; same property as is_moderated) + is_moderated: Whether the space is moderated (optional; same property as is_locked) is_public: Whether the space is public (optional) - is_announcement_only: Whether the space is announcement-only (only moderators can post) (optional) + is_announcement_only: Whether only moderators can post (optional) description: Description of the space (optional) - + Returns: - Dictionary containing the created space details + Standardized response dictionary with success/error information """ - # Call the underlying room function result = create_webex_room( - title=title, - team_id=team_id, - classification_id=classification_id, - is_locked=is_locked, - is_moderated=is_moderated, - is_public=is_public, - is_announcement_only=is_announcement_only, - description=description + title=title, team_id=team_id, classification_id=classification_id, + is_locked=is_locked, is_moderated=is_moderated, is_public=is_public, + is_announcement_only=is_announcement_only, description=description ) - - # Update the response to use "space" terminology - if result.get('success'): - result['space'] = result.pop('room', None) - + if result.get('success') and 'data' in result: + result['data']['space'] = result['data'].pop('room', None) return result @@ -426,60 +330,57 @@ def update_webex_space( ) -> Dict[str, Any]: """ Update an existing Webex space. - Note: This is an alias for update_webex_room - "room" and "space" are synonymous in Webex. - + Note: This is an alias for update_webex_room — "room" and "space" are synonymous in Webex. + Args: space_id: Space ID to update (required) title: New title for the space (optional) classification_id: New classification for the space (optional) - is_locked: Whether the space should be locked (optional, same as is_moderated) - is_moderated: Whether the space should be moderated (optional, same as is_locked) + is_locked: Whether the space should be locked (optional; same property as is_moderated) + is_moderated: Whether the space should be moderated (optional; same property as is_locked) is_public: Whether the space should be public (optional) - is_announcement_only: Whether the space should be announcement-only (only moderators can post) (optional) + is_announcement_only: Whether only moderators can post (optional) description: New description for the space (optional) - + Returns: - Dictionary containing the updated space details + Standardized response dictionary with success/error information """ - # Call the underlying room function with mapped parameters result = update_webex_room( - room_id=space_id, # Map space_id to room_id - title=title, - classification_id=classification_id, - is_locked=is_locked, - is_moderated=is_moderated, - is_public=is_public, - is_announcement_only=is_announcement_only, - description=description + room_id=space_id, title=title, classification_id=classification_id, + is_locked=is_locked, is_moderated=is_moderated, is_public=is_public, + is_announcement_only=is_announcement_only, description=description ) - - # Update the response to use "space" terminology - if result.get('success'): - result['space'] = result.pop('room', None) - result['space_id'] = result.pop('room_id', None) - + if result.get('success') and 'data' in result: + result['data']['space'] = result['data'].pop('room', None) return result -def get_webex_space( - space_id: str -) -> Dict[str, Any]: +def get_webex_space(space_id: str) -> Dict[str, Any]: """ Get detailed information about a specific Webex space. - Note: This is an alias for get_webex_room - "room" and "space" are synonymous in Webex. - + Note: This is an alias for get_webex_room — "room" and "space" are synonymous in Webex. + Args: space_id: Space ID to get details for (required) - + Returns: - Dictionary containing the space details + Standardized response dictionary with success/error information """ - # Call the underlying room function with mapped parameters - result = get_webex_room(room_id=space_id) # Map space_id to room_id - - # Update the response to use "space" terminology - if result.get('success'): - result['space'] = result.pop('room', None) - result['space_id'] = result.pop('room_id', None) - + result = get_webex_room(room_id=space_id) + if result.get('success') and 'data' in result: + result['data']['space'] = result['data'].pop('room', None) return result + + +def delete_webex_space(space_id: str) -> Dict[str, Any]: + """ + Delete a Webex space. + Note: This is an alias for delete_webex_room — "room" and "space" are synonymous in Webex. + + Args: + space_id: Space ID to delete (required) + + Returns: + Standardized response dictionary with success/error information + """ + return delete_webex_room(room_id=space_id)